You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/07/09 09:28:20 UTC
[GitHub] [flink] lirui-apache commented on a change in pull request #16416: [FLINK-22657][Connectors/Hive] HiveParserDDLSemanticAnalyzer return operations directly
lirui-apache commented on a change in pull request #16416:
URL: https://github.com/apache/flink/pull/16416#discussion_r666656070
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
##########
@@ -1388,23 +1723,41 @@ private Serializable analyzeAlterTableRenameCol(
}
String tblName = HiveParserBaseSemanticAnalyzer.getDotName(qualified);
- return HiveParserAlterTableDesc.changeColumn(
- tblName,
- HiveParserBaseSemanticAnalyzer.unescapeIdentifier(oldColName),
- HiveParserBaseSemanticAnalyzer.unescapeIdentifier(newColName),
- newType,
- newComment,
- first,
- flagCol,
- isCascade);
+
+ ObjectIdentifier tableIdentifier = parseObjectIdentifier(tblName);
+ CatalogBaseTable catalogBaseTable = getCatalogBaseTable(tableIdentifier);
+ if (catalogBaseTable instanceof CatalogView) {
Review comment:
Can we do this check in the outer method so that we don't have to do this in each alter table methods?
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
##########
@@ -1441,16 +1853,24 @@ private Serializable analyzeAlterTableDropParts(
validateAlterTableType(tab, AlterTableDesc.AlterTableTypes.DROPPARTITION, expectView);
- return new DropPartitionDesc(qualified[0], qualified[1], partSpecs, ifExists);
+ ObjectIdentifier tableIdentifier =
+ catalogManager.qualifyIdentifier(
+ UnresolvedIdentifier.of(qualified[0], qualified[1]));
+ CatalogBaseTable catalogBaseTable = getCatalogBaseTable(tableIdentifier);
+ if (catalogBaseTable instanceof CatalogView) {
+ throw new ValidationException("DROP PARTITION for a view is not supported");
+ }
+ List<CatalogPartitionSpec> specs =
+ partSpecs.stream().map(CatalogPartitionSpec::new).collect(Collectors.toList());
+ return new DropPartitionsOperation(tableIdentifier, ifExists, specs);
}
/**
* Add one or more partitions to a table. Useful when the data has been copied to the right
* location by some other process.
*/
- private Serializable analyzeAlterTableAddParts(
- String[] qualified, CommonTree ast, boolean expectView) throws SemanticException {
-
+ private Operation convertAlterTableAddParts(
+ String[] qualified, CommonTree ast, boolean expectView) {
Review comment:
We don't support partitions for views. So let's remove the `expectView` parameter.
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
##########
@@ -841,25 +1141,36 @@ private Serializable analyzeDropDatabase(HiveParserASTNode ast) {
ifCascade = true;
}
- return new HiveParserDropDatabaseDesc(dbName, ifExists, ifCascade);
+ return new DropDatabaseOperation(
+ catalogManager.getCurrentCatalog(), dbName, ifExists, ifCascade);
}
- private Serializable analyzeSwitchDatabase(HiveParserASTNode ast) {
+ private Operation convertSwitchDatabase(HiveParserASTNode ast) {
String dbName =
HiveParserBaseSemanticAnalyzer.unescapeIdentifier(ast.getChild(0).getText());
- SwitchDatabaseDesc switchDatabaseDesc = new SwitchDatabaseDesc(dbName);
- return new DDLWork(new HashSet<>(), new HashSet<>(), switchDatabaseDesc);
+ return new UseDatabaseOperation(catalogManager.getCurrentCatalog(), dbName);
}
- private Serializable analyzeDropTable(HiveParserASTNode ast, TableType expectedType) {
+ private Operation convertDropTable(HiveParserASTNode ast, TableType expectedType) {
String tableName =
HiveParserBaseSemanticAnalyzer.getUnescapedName(
(HiveParserASTNode) ast.getChild(0));
boolean ifExists = (ast.getFirstChildWithType(HiveASTParser.TOK_IFEXISTS) != null);
- boolean ifPurge = (ast.getFirstChildWithType(HiveASTParser.KW_PURGE) != null);
- return new HiveParserDropTableDesc(
- tableName, expectedType == TableType.VIRTUAL_VIEW, ifExists, ifPurge);
+ ObjectIdentifier identifier = parseObjectIdentifier(tableName);
+ CatalogBaseTable baseTable = getCatalogBaseTable(identifier, true);
+
+ if (expectedType == TableType.VIRTUAL_VIEW) {
+ if (baseTable instanceof CatalogTable) {
+ throw new ValidationException("DROP VIEW for a table is not allowed");
+ }
+ return new DropViewOperation(identifier, ifExists, false);
+ } else {
+ if (baseTable instanceof CatalogView) {
+ throw new ValidationException("DROP TABLE for a view is not allowed");
+ }
+ return new DropTableOperation(identifier, ifExists, false);
+ }
}
private void validateAlterTableType(
Review comment:
This method can be simplified and only checks whether a table is non-native
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
##########
@@ -1414,10 +1767,69 @@ private Serializable analyzeAlterTableModifyCols(
isCascade = true;
}
- return HiveParserAlterTableDesc.addReplaceColumns(tblName, newCols, replace, isCascade);
+ ObjectIdentifier tableIdentifier = parseObjectIdentifier(tblName);
+ CatalogBaseTable catalogBaseTable = getCatalogBaseTable(tableIdentifier);
+ if (catalogBaseTable instanceof CatalogView) {
+ throw new ValidationException("ALTER TABLE for a view is not allowed");
+ }
+ CatalogTable oldTable = (CatalogTable) catalogBaseTable;
+
+ // prepare properties
+ Map<String, String> props = new HashMap<>(oldTable.getOptions());
+ props.put(ALTER_TABLE_OP, ALTER_COLUMNS.name());
+ if (isCascade) {
+ props.put(ALTER_COL_CASCADE, "true");
+ }
+ TableSchema oldSchema = oldTable.getSchema();
+ final int numPartCol = oldTable.getPartitionKeys().size();
+ TableSchema.Builder builder = TableSchema.builder();
+ // add existing non-part col if we're not replacing
+ if (!replace) {
+ List<TableColumn> nonPartCols =
+ oldSchema.getTableColumns().subList(0, oldSchema.getFieldCount() - numPartCol);
+ for (TableColumn column : nonPartCols) {
+ builder.add(column);
+ }
+ setWatermarkAndPK(builder, oldSchema);
+ }
+ // add new cols
+ for (FieldSchema col : newCols) {
+ builder.add(
+ TableColumn.physical(
+ col.getName(),
+ HiveTypeUtil.toFlinkType(
+ TypeInfoUtils.getTypeInfoFromTypeString(col.getType()))));
+ }
+ // add part cols
+ List<TableColumn> partCols =
+ oldSchema
+ .getTableColumns()
+ .subList(oldSchema.getFieldCount() - numPartCol, oldSchema.getFieldCount());
+ for (TableColumn column : partCols) {
+ builder.add(column);
+ }
+ return new AlterTableSchemaOperation(
+ tableIdentifier,
+ new CatalogTableImpl(
+ builder.build(),
+ oldTable.getPartitionKeys(),
+ props,
+ oldTable.getComment()));
+ }
+
+ private static void setWatermarkAndPK(TableSchema.Builder builder, TableSchema schema) {
+ for (WatermarkSpec watermarkSpec : schema.getWatermarkSpecs()) {
+ builder.watermark(watermarkSpec);
+ }
+ schema.getPrimaryKey()
+ .ifPresent(
+ pk -> {
+ builder.primaryKey(
+ pk.getName(), pk.getColumns().toArray(new String[0]));
+ });
}
- private Serializable analyzeAlterTableDropParts(
+ private Operation convertAlterTableDropParts(
Review comment:
Same here, remove `expectView`
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
##########
@@ -1495,25 +1920,96 @@ private Serializable analyzeAlterTableAddParts(
// add the last one
if (currentPart != null) {
- addPartitionDesc.addPartition(currentPart, currentLocation);
+ specs.add(new CatalogPartitionSpec(currentPart));
+ Map<String, String> props = new HashMap<>();
+ if (currentLocation != null) {
+ props.put(TABLE_LOCATION_URI, currentLocation);
+ }
+ partitions.add(new CatalogPartitionImpl(props, null));
}
- return new DDLWork(getInputs(), getOutputs(), addPartitionDesc);
+ ObjectIdentifier tableIdentifier =
+ tab.getDbName() == null
+ ? parseObjectIdentifier(tab.getTableName())
+ : catalogManager.qualifyIdentifier(
+ UnresolvedIdentifier.of(tab.getDbName(), tab.getTableName()));
+ CatalogBaseTable catalogBaseTable = getCatalogBaseTable(tableIdentifier);
+ if (catalogBaseTable instanceof CatalogView) {
Review comment:
No need to do the check here. If it's a view, we should have thrown "alter table for view is not allowed"
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
##########
@@ -1495,25 +1920,96 @@ private Serializable analyzeAlterTableAddParts(
// add the last one
if (currentPart != null) {
- addPartitionDesc.addPartition(currentPart, currentLocation);
+ specs.add(new CatalogPartitionSpec(currentPart));
+ Map<String, String> props = new HashMap<>();
+ if (currentLocation != null) {
+ props.put(TABLE_LOCATION_URI, currentLocation);
+ }
+ partitions.add(new CatalogPartitionImpl(props, null));
}
- return new DDLWork(getInputs(), getOutputs(), addPartitionDesc);
+ ObjectIdentifier tableIdentifier =
+ tab.getDbName() == null
Review comment:
we can just get DB and table name from `qualified` parameter?
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
##########
@@ -1285,59 +1647,32 @@ private Serializable analyzeShowTables(HiveParserASTNode ast, boolean expectView
*
* @param ast The parsed command tree.
*/
- private Serializable analyzeShowFunctions(HiveParserASTNode ast) {
- ShowFunctionsDesc showFuncsDesc;
- if (ast.getChildCount() == 1) {
- String funcNames =
- HiveParserBaseSemanticAnalyzer.stripQuotes(ast.getChild(0).getText());
- showFuncsDesc = new ShowFunctionsDesc(getResFile(), funcNames);
- } else if (ast.getChildCount() == 2) {
+ private Operation convertShowFunctions(HiveParserASTNode ast) {
+ if (ast.getChildCount() == 2) {
assert (ast.getChild(0).getType() == HiveASTParser.KW_LIKE);
throw new ValidationException("SHOW FUNCTIONS LIKE is not supported yet");
- } else {
- showFuncsDesc = new ShowFunctionsDesc(getResFile());
- }
- return new DDLWork(getInputs(), getOutputs(), showFuncsDesc);
- }
-
- /**
- * Add the task according to the parsed command tree. This is used for the CLI command "DESCRIBE
- * FUNCTION;".
- *
- * @param ast The parsed command tree.
- */
- private Serializable analyzeDescFunction(HiveParserASTNode ast) {
- String funcName;
- boolean isExtended;
-
- if (ast.getChildCount() == 1) {
- funcName = HiveParserBaseSemanticAnalyzer.stripQuotes(ast.getChild(0).getText());
- isExtended = false;
- } else if (ast.getChildCount() == 2) {
- funcName = HiveParserBaseSemanticAnalyzer.stripQuotes(ast.getChild(0).getText());
- isExtended = true;
- } else {
- throw new ValidationException("Unexpected Tokens at DESCRIBE FUNCTION");
}
-
- DescFunctionDesc descFuncDesc = new DescFunctionDesc(getResFile(), funcName, isExtended);
- return new DDLWork(getInputs(), getOutputs(), descFuncDesc);
+ return new ShowFunctionsOperation();
}
- private Serializable analyzeAlterTableRename(
+ private Operation convertAlterTableRename(
String[] source, HiveParserASTNode ast, boolean expectView) throws SemanticException {
String[] target =
HiveParserBaseSemanticAnalyzer.getQualifiedTableName(
(HiveParserASTNode) ast.getChild(0));
String sourceName = HiveParserBaseSemanticAnalyzer.getDotName(source);
String targetName = HiveParserBaseSemanticAnalyzer.getDotName(target);
+ ObjectIdentifier objectIdentifier = parseObjectIdentifier(sourceName);
+ checkAlterTableLegal(objectIdentifier, expectView);
- return HiveParserAlterTableDesc.rename(sourceName, targetName, expectView);
+ return expectView
+ ? new AlterViewRenameOperation(objectIdentifier, parseObjectIdentifier(targetName))
+ : new AlterTableRenameOperation(
+ objectIdentifier, parseObjectIdentifier(targetName));
}
- private Serializable analyzeAlterTableRenameCol(
- String[] qualified, HiveParserASTNode ast, HashMap<String, String> partSpec)
+ private Operation convertAlterTableRenameCol(String[] qualified, HiveParserASTNode ast)
Review comment:
It does more than just rename. So let's call it `convertAlterTableChangeCol`
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
##########
@@ -716,29 +836,188 @@ private Serializable analyzeCreateTable(HiveParserASTNode ast) throws SemanticEx
case ctas: // create table as select
tblProps = addDefaultProperties(tblProps);
- HiveParserCreateTableDesc createTableDesc =
- new HiveParserCreateTableDesc(
- dbDotTab,
- isExt,
- ifNotExists,
- isTemporary,
- cols,
- partCols,
- comment,
- location,
- tblProps,
- rowFormatParams,
- storageFormat,
- primaryKeys,
- notNulls);
- return new CreateTableASDesc(createTableDesc, selectStmt);
+ // analyze the query
+ HiveParserCalcitePlanner calcitePlanner =
+ hiveParser.createCalcitePlanner(context, queryState, hiveShim);
+ calcitePlanner.setCtasCols(cols);
+ RelNode queryRelNode = calcitePlanner.genLogicalPlan(selectStmt);
+ // create a table to represent the dest table
+ String[] dbTblName = dbDotTab.split("\\.");
+ Table destTable = new Table(Table.getEmptyTable(dbTblName[0], dbTblName[1]));
+ destTable.getSd().setCols(cols);
+ // create the insert operation
+ CatalogSinkModifyOperation insertOperation =
+ dmlHelper.createInsertOperation(
+ queryRelNode,
+ destTable,
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ false);
+
+ CreateTableOperation createTableOperation =
+ (CreateTableOperation)
Review comment:
Make `convertCreateTable` return `CreateTableOperation`, no need to do the cast
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
##########
@@ -449,15 +527,24 @@ private Serializable analyzerCreateFunction(HiveParserASTNode ast) {
"Temporary function cannot be created with a qualified name.");
}
- CreateFunctionDesc desc = new CreateFunctionDesc();
- desc.setFunctionName(functionName);
- desc.setTemp(isTemporaryFunction);
- desc.setClassName(className);
- desc.setResources(Collections.emptyList());
- return new FunctionWork(desc);
+ if (isTemporaryFunction) {
+ // hive's temporary function is more like flink's temp system function, e.g. doesn't
+ // belong to a catalog/db
+ // the DDL analyzer makes sure temp function name is not a compound one
Review comment:
this comment should be deleted
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
##########
@@ -1466,14 +1886,19 @@ private Serializable analyzeAlterTableAddParts(
Map<String, String> currentPart = null;
Review comment:
rename it to `currentSpec`
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
##########
@@ -295,52 +375,46 @@ public Serializable analyzeInternal(HiveParserASTNode input) throws SemanticExce
break;
}
case HiveASTParser.TOK_DROPTABLE:
- res = analyzeDropTable(ast, null);
+ res = convertDropTable(ast, null);
break;
case HiveASTParser.TOK_DESCTABLE:
- res = analyzeDescribeTable(ast);
+ res = convertDescribeTable(ast);
break;
case HiveASTParser.TOK_SHOWDATABASES:
- res = analyzeShowDatabases(ast);
+ res = convertShowDatabases();
break;
case HiveASTParser.TOK_SHOWTABLES:
- res = analyzeShowTables(ast, false);
+ res = convertShowTables(ast, false);
break;
case HiveASTParser.TOK_SHOWFUNCTIONS:
- res = analyzeShowFunctions(ast);
+ res = convertShowFunctions(ast);
break;
case HiveASTParser.TOK_SHOWVIEWS:
- res = analyzeShowTables(ast, true);
- break;
- case HiveASTParser.TOK_DESCFUNCTION:
- res = analyzeDescFunction(ast);
- break;
- case HiveASTParser.TOK_DESCDATABASE:
- res = analyzeDescDatabase(ast);
+ res = convertShowTables(ast, true);
break;
case HiveASTParser.TOK_DROPVIEW:
- res = analyzeDropTable(ast, TableType.VIRTUAL_VIEW);
+ res = convertDropTable(ast, TableType.VIRTUAL_VIEW);
break;
case HiveASTParser.TOK_ALTERVIEW:
Review comment:
extract this into a separate method
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
##########
@@ -1224,23 +1590,19 @@ private Serializable analyzeShowPartitions(HiveParserASTNode ast) throws Semanti
validateTable(tableName, null);
Review comment:
We don't really need this method. Catalog will check whether the table exists when the operation is executed.
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
##########
@@ -197,18 +287,9 @@ private ObjectPath toObjectPath(String name) throws SemanticException {
return new ObjectPath(parts[0], parts[1]);
}
- private HashSet<ReadEntity> getInputs() {
- return new HashSet<>();
- }
-
- private HashSet<WriteEntity> getOutputs() {
- return new HashSet<>();
- }
-
- public Serializable analyzeInternal(HiveParserASTNode input) throws SemanticException {
-
+ public Operation convertToOperation(HiveParserASTNode input) throws SemanticException {
HiveParserASTNode ast = input;
- Serializable res = null;
+ Operation res = null;
switch (ast.getType()) {
case HiveASTParser.TOK_ALTERTABLE:
Review comment:
Let's extract this into a separate method to handle all the ALTER TABLE operations
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org