@Test public void testCreateTableWithMinusInOptionKey() { final String sql = "create table source_table(\n" + " a int,\n" + " b bigint,\n" + " c varchar\n" + ") with (\n" + " 'a-b-c-d124' = 'ab',\n" + " 'a.b-c-d.e-f.g' = 'ada',\n" + " 'a.b-c-d.e-f1231.g' = 'ada',\n" + " 'a.b-c-d.*' = 'adad')\n"; final FlinkPlannerImpl planner = getPlannerBySqlDialect(SqlDialect.DEFAULT); SqlNode node = planner.parse(sql); assert node instanceof SqlCreateTable; Operation operation = SqlToOperationConverter.convert(planner, node); assert operation instanceof CreateTableOperation; CreateTableOperation op = (CreateTableOperation) operation; CatalogTable catalogTable = op.getCatalogTable(); Map<String, String> properties = catalogTable.toProperties() .entrySet().stream() .filter(e -> !e.getKey().contains("schema")) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); Map<String, String> sortedProperties = new TreeMap<>(properties); final String expected = "{a-b-c-d124=ab, " + "a.b-c-d.*=adad, " + "a.b-c-d.e-f.g=ada, " + "a.b-c-d.e-f1231.g=ada}"; assertEquals(expected, sortedProperties.toString()); }
public static Operation convert(FlinkPlannerImpl flinkPlanner, SqlNode sqlNode) { final SqlNode validated = flinkPlanner.validate(sqlNode); SqlToOperationConverter converter = new SqlToOperationConverter(flinkPlanner); if (validated instanceof SqlCreateTable) { return converter.convertCreateTable((SqlCreateTable) validated); } if (validated instanceof SqlDropTable) { return converter.convertDropTable((SqlDropTable) validated); } else if (validated instanceof RichSqlInsert) { return converter.convertSqlInsert((RichSqlInsert) validated); } else if (validated.getKind().belongsTo(SqlKind.QUERY)) { return converter.convertSqlQuery(validated); } else { throw new TableException("Unsupported node type " + validated.getClass().getSimpleName()); } }
SqlToOperationConverter { public static Operation convert(FlinkPlannerImpl flinkPlanner, SqlNode sqlNode) { final SqlNode validated = flinkPlanner.validate(sqlNode); SqlToOperationConverter converter = new SqlToOperationConverter(flinkPlanner); if (validated instanceof SqlCreateTable) { return converter.convertCreateTable((SqlCreateTable) validated); } if (validated instanceof SqlDropTable) { return converter.convertDropTable((SqlDropTable) validated); } else if (validated instanceof RichSqlInsert) { return converter.convertSqlInsert((RichSqlInsert) validated); } else if (validated.getKind().belongsTo(SqlKind.QUERY)) { return converter.convertSqlQuery(validated); } else { throw new TableException("Unsupported node type " + validated.getClass().getSimpleName()); } } }
SqlToOperationConverter { public static Operation convert(FlinkPlannerImpl flinkPlanner, SqlNode sqlNode) { final SqlNode validated = flinkPlanner.validate(sqlNode); SqlToOperationConverter converter = new SqlToOperationConverter(flinkPlanner); if (validated instanceof SqlCreateTable) { return converter.convertCreateTable((SqlCreateTable) validated); } if (validated instanceof SqlDropTable) { return converter.convertDropTable((SqlDropTable) validated); } else if (validated instanceof RichSqlInsert) { return converter.convertSqlInsert((RichSqlInsert) validated); } else if (validated.getKind().belongsTo(SqlKind.QUERY)) { return converter.convertSqlQuery(validated); } else { throw new TableException("Unsupported node type " + validated.getClass().getSimpleName()); } } private  SqlToOperationConverter(FlinkPlannerImpl flinkPlanner); }
SqlToOperationConverter { public static Operation convert(FlinkPlannerImpl flinkPlanner, SqlNode sqlNode) { final SqlNode validated = flinkPlanner.validate(sqlNode); SqlToOperationConverter converter = new SqlToOperationConverter(flinkPlanner); if (validated instanceof SqlCreateTable) { return converter.convertCreateTable((SqlCreateTable) validated); } if (validated instanceof SqlDropTable) { return converter.convertDropTable((SqlDropTable) validated); } else if (validated instanceof RichSqlInsert) { return converter.convertSqlInsert((RichSqlInsert) validated); } else if (validated.getKind().belongsTo(SqlKind.QUERY)) { return converter.convertSqlQuery(validated); } else { throw new TableException("Unsupported node type " + validated.getClass().getSimpleName()); } } private  SqlToOperationConverter(FlinkPlannerImpl flinkPlanner); static Operation convert(FlinkPlannerImpl flinkPlanner, SqlNode sqlNode); }
SqlToOperationConverter { public static Operation convert(FlinkPlannerImpl flinkPlanner, SqlNode sqlNode) { final SqlNode validated = flinkPlanner.validate(sqlNode); SqlToOperationConverter converter = new SqlToOperationConverter(flinkPlanner); if (validated instanceof SqlCreateTable) { return converter.convertCreateTable((SqlCreateTable) validated); } if (validated instanceof SqlDropTable) { return converter.convertDropTable((SqlDropTable) validated); } else if (validated instanceof RichSqlInsert) { return converter.convertSqlInsert((RichSqlInsert) validated); } else if (validated.getKind().belongsTo(SqlKind.QUERY)) { return converter.convertSqlQuery(validated); } else { throw new TableException("Unsupported node type " + validated.getClass().getSimpleName()); } } private  SqlToOperationConverter(FlinkPlannerImpl flinkPlanner); static Operation convert(FlinkPlannerImpl flinkPlanner, SqlNode sqlNode);  }