Peter Rozsa has posted comments on this change. ( http://gerrit.cloudera.org:8080/21423 )
Change subject: IMPALA-12732: Add support for MERGE statements for Iceberg tables ...................................................................... Patch Set 7: (63 comments) Thank you Zoltan and Gabor! I covered most of the comments, some of them are still open. If we could resolve those, I'll split up this change into multiple pieces to make reviewing easier. http://gerrit.cloudera.org:8080/#/c/21423/6//COMMIT_MSG Commit Message: http://gerrit.cloudera.org:8080/#/c/21423/6//COMMIT_MSG@7 PS6, Line 7: IMPALA-12732: Add support for MERGE statements for Iceberg tables > I know MERGE is a known SQL functionality but I'd find it useful to add a f Done http://gerrit.cloudera.org:8080/#/c/21423/6//COMMIT_MSG@11 PS6, Line 11: ource table. This change adds MERGE > Would you mind adding some examples? Done http://gerrit.cloudera.org:8080/#/c/21423/6//COMMIT_MSG@24 PS6, Line 24: > nit: maybe 'phase' is a better word here Done http://gerrit.cloudera.org:8080/#/c/21423/6//COMMIT_MSG@47 PS6, Line 47: > We should also have authorization tests. Also for fine-grained authz, e.g. Done http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-node.h File be/src/exec/iceberg-merge-node.h: http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-node.h@18 PS6, Line 18: #pragma once : > nit: we prefer #pragma once Done http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-node.h@43 PS6, Line 43: /// for merge action and for the target table. > Please add class comment Done http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-node.h@101 PS6, Line 101: /// added to the > We prefer std::unique_ptr Done http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-node.h@158 PS6, Line 158: se(con > no need for 'inline', member-functions defined inside class definitions are Done http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-node.cc File be/src/exec/iceberg-merge-node.cc: http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-node.cc@165 PS6, Line 165: os > nit: too much indent Done http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-node.cc@178 PS6, Line 178: auto row = iter.Ge > I'm not sure if we want to reset last_row_ each time EvaluateCases() is inv Removed http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-node.cc@186 PS6, Line 186: last_row_ = row; : > Can we also output the duplicated row with PrintTuple()? It's a tricky situation; it's easy to print the target table's tuple, but it also contains the virtual columns, so for small tables, the path will be the dominant part of the error message; I think it's acceptable. To print the source tuples, we have to consider cases where the source row consists of multiple tuples, so first we have to distinguish the source tuples or the tuple that is used for the join, and then print them. There's another quite simple option; using PrintRow. What do you think? Examples: 1. ERROR: Duplicate row found: one target table row matched more than one source row. Target row: (3 3 3 hdfs://localhost:20500/test-warehouse/target_part/data/514806418e9554cc-654b331700000005_831375517_data.0.parq 3) 2. ERROR: Duplicate row found: one target table row matched more than one source row. Target row: (4 4 4 hdfs://localhost:20500/test-warehouse/target_part/data/514806418e9554cc-654b331700000005_831375517_data.0.parq 4), Source row: (4 4 4) (4 4 4) (4 4 4) (50) 3. ERROR: Duplicate row found: one target table row matched more than one source row. Affected row: [(4 4 4 hdfs://localhost:20500/test-warehouse/target_part/data/514806418e9554cc-654b331700000005_831375517_data.0.parq 0) (4 4 4) (4 4 4) (4 4 4) (50)] // The source table ref here are 4 joined tables http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-node.cc@209 PS6, Line 209: if (ReachedLimit() | > nit: to reduce nesting, we could have Done http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-node.cc@210 PS6, Line 210: } : return Status::OK(); : } : : void IcebergMergeNode::AddRow( : RowBatch* output_batch, IcebergMergeCase* merge_case, TupleRow* row) { : int dst_row_idx = output_batch->AddRow(); : TupleRow* dst_row = output_batch->GetRow(dst_row_idx); : : auto* target_tuple = : Tuple::Create(row_descriptor_.tuple_descriptors()[target_tuple_idx_]->byte_size(), : output_batch->tuple_data_pool()); : auto* merge_action_tuple = Tuple::Create( : row_descriptor_.tuple_descriptors()[merge_action_tuple_idx_]->byte_size(), : output_batch->tuple_data_pool()); : : TIcebergMergeSinkAction::type action = merge_case->SinkAction(); : : dst_row->SetTuple(target_tuple_idx_, target_tuple); : dst_row->SetTuple(merge_action_tuple_idx_, merge_action_tuple); : : for (int i = 0; i < row_descriptor_.tuple_descriptors().size(); i++) { : if (i != target_tuple_idx_ && i != merge_action_tuple_idx_) { : dst_row->SetTuple(i, nullptr); : } : } : : target_tuple->MaterializeExprs<false, false>(row, : *row_descriptor_.tuple_descriptors()[target_tuple_idx_], : merge_case->combined_evaluat > nit: for readability, this could go to its own member function Done http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-node.cc@248 PS6, Line 248: bool IcebergMergeNode::IsDuplicateRow( : TupleRow* previous_row, TupleRow* actual_row, int target_tuple_idx) { : i > nit: fits single line Done http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-node.cc@252 PS6, Line 252: if (previous_row_target_tuple == nullptr) { return false; } : auto actual_row_target_tuple = actual_row->GetTuple(target_tuple_idx); : r > nit: fits single line Done http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-node.cc@262 PS6, Line 262: return ExecNode::Reset(state, row_batch); > last_row_ could be nulled out. Done http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-sink.h File be/src/exec/iceberg-merge-sink.h: http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-sink.h@18 PS6, Line 18: #pragma once : > nit: please use #pragma once Done http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-sink.h@22 PS6, Line 22: namespace impala { > nit: please add blank line Done http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-sink.h@38 PS6, Line 38: const IcebergMergeSink > nit: this is usually const T& Done http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-sink.h@40 PS6, Line 40: const IcebergMergeSink > nit: missing const here as well Done http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-sink.cc File be/src/exec/iceberg-merge-sink.cc: http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/exec/iceberg-merge-sink.cc@110 PS6, Line 110: delete_rows.CopyRow(row, output_row); : delete_rows.CommitLastRow(); > Do we really need to DeepCopy the incoming rows? Setting the pointers isn't Replaced with CopyRow http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/service/client-request-state.cc File be/src/service/client-request-state.cc: http://gerrit.cloudera.org:8080/#/c/21423/6/be/src/service/client-request-state.cc@1671 PS6, Line 1671: > nit: requires 4 spaces indent Done http://gerrit.cloudera.org:8080/#/c/21423/6/common/thrift/PlanNodes.thrift File common/thrift/PlanNodes.thrift: http://gerrit.cloudera.org:8080/#/c/21423/6/common/thrift/PlanNodes.thrift@734 PS6, Line 734: // Output expressions that will transform the values of the incoming : // merge result set. : 1: required list<Exprs.TExpr> o > Comments for the fields would be helpful Done http://gerrit.cloudera.org:8080/#/c/21423/6/common/thrift/PlanNodes.thrift@789 PS6, Line 789: 18: optional TExchangeNode exchange_node > field ids don't need to be in order. Reassigning existing field ids can mak Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/cup/sql-parser.cup File fe/src/main/cup/sql-parser.cup: http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/cup/sql-parser.cup@1020 PS6, Line 1020: table_ref:source > Can we also allow queries here (in a follow-up patch)? Table refs can be inline views so queries like `merge into target using (select * from source union select 1,2,3) source on target.id = source.id` are possible http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/DmlStatementBase.java File fe/src/main/java/org/apache/impala/analysis/DmlStatementBase.java: http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/DmlStatementBase.java@105 PS6, Line 105: > nit: unnecessary line break Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/DmlStatementBase.java@131 PS6, Line 131: > nit: unnecessary line break Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/DmlStatementBase.java@139 PS6, Line 139: > nit: unnecessary line break Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java File fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java: http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java@32 PS6, Line 32: ai > nit: please import individual classes Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java@41 PS6, Line 41: Util; > nit: could be all capital Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java@91 PS6, Line 91: > We should check that the target is indeed an Iceberg table, otherwise we sh It's checked in MergeStmt, I'll add a precondition check here. http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java@103 PS6, Line 103: Preconditions.checkState(table instanceof FeIcebergTable); : icebergTable_ = (FeIcebergTable) table; : for (Column column : icebergTable_.getColumns()) { : Path slotPath = new Path(targetTableRef_.desc_, : Collections.singletonList(column.getName())); : slotPath.resolve(); : analyzer.registerSlotRef(slotPath); : } : sourceTableRef_.analyze(analyzer); : : queryStmt_ = prepareQuery(); : queryStmt_.analyze(analyzer); : > The merge action related parts could be moved to a helper method for readab Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java@136 PS6, Line 136: } : : private void addMergeActionTuple(Analyzer analyzer) { : mergeActionTuple_ = : analyzer.getDescTbl().createTupleDescriptor(MERGE_ACTION_TUPLE_NAME); : : SlotDescriptor sd = analyzer.addSlotDescriptor(mergeActionTuple_); : sd.setType(Type.TINYINT); : sd.setIsMaterialized(true); : sd.setIsNullable(false); : : Expr mergeActionExpr = new SlotRef(sd); : sd.setSourceExpr(mergeActionExpr); : expressions_.mergeActionExpression(mergeActionExpr); : } : : @Override : public void substituteResultExprs(ExprSubstitutionMap smap, : > nit: can we move it next to getPlanNode()? Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java@159 PS6, Line 159: } : : @Override : public List<Expr> getResultExprs() { : return expressions_.resultExpressions(); : } : : @Override : public List<Expr> getPartitionKeyExprs() { : return expressions_.targetPartitionExpressions(); : } > Could be placed to a method in MergeExpressions. Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java@240 PS6, Line 240: > nit: missing space before * Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java@240 PS6, Line 240: > Isn't it TINYINT? Corrected http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java@440 PS6, Line 440: return targetSorting.order_; : } : > please move next to the getter Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java@444 PS6, Line 444: targetSorting = sorting; : } : > Please move next to the setter Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/MergeCase.java File fe/src/main/java/org/apache/impala/analysis/MergeCase.java: http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/MergeCase.java@33 PS6, Line 33: filte > nit: redundant 'this.'. Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/MergeCase.java@40 PS6, Line 40: filte > nit: redundant 'this.' Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/MergeCase.java@47 PS6, Line 47: if (isAnalyzed()) return; > nit: you could follow the usual pattern: Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/MergeInsert.java File fe/src/main/java/org/apache/impala/analysis/MergeInsert.java: http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/MergeInsert.java@35 PS6, Line 35: > Could you please add comment? Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/MergeInsert.java@83 PS6, Line 83: > Could you pelase add comment to this method? Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/MergeInsert.java@142 PS6, Line 142: mnMatch(column > Could you please add a comment for this method? Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/MergeStmt.java File fe/src/main/java/org/apache/impala/analysis/MergeStmt.java: http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/MergeStmt.java@48 PS6, Line 48: @Override > nit: You could follow the usual pattern: Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/MergeStmt.java@68 PS6, Line 68: } > nit: redundant empty line Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/analysis/MergeStmt.java@78 PS6, Line 78: blRefs.add(sourceTableRef_); > This could be a precondition This condition is required to avoid the case where an inline view is provided as a target, as the table loader will fail on inline views. Later, in analyze, the error is reported properly. http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/planner/IcebergMergeNode.java File fe/src/main/java/org/apache/impala/planner/IcebergMergeNode.java: http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/planner/IcebergMergeNode.java@129 PS6, Line 129: } > Can this statement have a limit? Removed. http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/planner/Planner.java File fe/src/main/java/org/apache/impala/planner/Planner.java: http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/planner/Planner.java@143 PS6, Line 143: singleNodePlan = addMergeNode(singleNodePlan, ctx_.getRootAnalyzer()); > Would it be possible to move it to L188? Or do you want it to be before val It's placed here to make sure the merge node is placed into one fragment with the join node. If it's placed after the distributed planner creates the fragments, the join node must be located, as it's behind the exchange and this step could be fragile. I feel that this is not a fortunate place to add the merge node, but the other way is more invasive and error-prone. http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/planner/Planner.java@228 PS6, Line 228: isUpdate() || ct > nit: this is the only place this method is being used. Could be isUpdateOrD Going with the second option. http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/service/IcebergCatalogOpExecutor.java File fe/src/main/java/org/apache/impala/service/IcebergCatalogOpExecutor.java: http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/service/IcebergCatalogOpExecutor.java@442 PS6, Line 442: up > indentation is off Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/util/IcebergUtil.java File fe/src/main/java/org/apache/impala/util/IcebergUtil.java: http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/main/java/org/apache/impala/util/IcebergUtil.java@1300 PS6, Line 1300: ; > Since PartitionExprBundle is basically a struct and these fields are public Done http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/test/java/org/apache/impala/analysis/AnalyzeModifyStmtsTest.java File fe/src/test/java/org/apache/impala/analysis/AnalyzeModifyStmtsTest.java: http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/test/java/org/apache/impala/analysis/AnalyzeModifyStmtsTest.java@264 PS6, Line 264: AnalyzesOk("merge into functional_parquet.iceberg_partitioned target " > for me this test seems identical to the one right above. Or do I miss somet Most probably I wanted to add a new case, and I duplicated the lines for it, but now I lost the context what should be covered here, I'll remove it, but I try to keep in mind that it was duplicated intentionally, to cover something similar. http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/test/java/org/apache/impala/analysis/AnalyzeModifyStmtsTest.java@384 PS6, Line 384: + "when not mat > Is this error msg introduced by this patch or did it exist before? I found It's the same semantics that InsertStmt uses, I followed that. http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/test/java/org/apache/impala/analysis/AnalyzeModifyStmtsTest.java@386 PS6, Line 386: "Column permutation me > Actually, this is also a 'More columns in VALUES' case similarly to above. Yes, the comment remained in by mistake, it's just another example for "More columns in VALUES" http://gerrit.cloudera.org:8080/#/c/21423/6/fe/src/test/java/org/apache/impala/analysis/AnalyzeModifyStmtsTest.java@410 PS6, Line 410: "Could not resolve column/field reference: 't.id'"); > nit: This test could go above the error cases. Also some comment about the Done http://gerrit.cloudera.org:8080/#/c/21423/6/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-merge.test File testdata/workloads/functional-planner/queries/PlannerTest/iceberg-merge.test: http://gerrit.cloudera.org:8080/#/c/21423/6/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-merge.test@5 PS6, Line 5: when matched then delete > We could use INNER JOIN for such MERGE statements that only have MATCHED cl Done http://gerrit.cloudera.org:8080/#/c/21423/6/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-merge.test@13 PS6, Line 13: result expressions: target.id, target.`user`, target.action, target.event_time > Do we need the result expressions when the type is DELETE? The merge node waits for the full row description at the evaluation, currently all rows are copied to the output row batch by evaluation the result expression of the matched case, it should be optimized to handle delete cases better, by only copying the virtual columns to the output batch, should I cover this case in this iteration or should I implement it in an additional patch? http://gerrit.cloudera.org:8080/#/c/21423/6/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-merge.test@223 PS6, Line 223: > I think this should be a PARTITIONED exchange on 'from_timestamp(target.eve Done http://gerrit.cloudera.org:8080/#/c/21423/6/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-merge.test@291 PS6, Line 291: > Again, this should be a PARTITIONED exchange. Done http://gerrit.cloudera.org:8080/#/c/21423/6/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-merge.test@324 PS6, Line 324: merge into functional_parquet.iceberg_partitioned target > I admit I haven't checked the code, but is there a precedence between these The order execution is based on the position of the cases; if an unconditional case precedes a conditional case, then the unconditional case will match first and execute. In this case, if we reverse the order of cases, all rows would be matched by the unconditional case. http://gerrit.cloudera.org:8080/#/c/21423/6/testdata/workloads/functional-query/queries/QueryTest/iceberg-merge.test File testdata/workloads/functional-query/queries/QueryTest/iceberg-merge.test: http://gerrit.cloudera.org:8080/#/c/21423/6/testdata/workloads/functional-query/queries/QueryTest/iceberg-merge.test@115 PS6, Line 115: ---- TYPES > Do you intentionally not asserting on some profile metrics from this test o Yes, the partitioned tables have different row statistics; it's broken down by partitions and it's harder to match. I was inspired by iceberg-update-basic.test. http://gerrit.cloudera.org:8080/#/c/21423/6/tests/query_test/test_iceberg.py File tests/query_test/test_iceberg.py: http://gerrit.cloudera.org:8080/#/c/21423/6/tests/query_test/test_iceberg.py@1791 PS6, Line 1791: f test_merge(self, vector, unique_database): > Was it intentional to add this line? Removed, it remained in accidentally after rebasing. -- To view, visit http://gerrit.cloudera.org:8080/21423 To unsubscribe, visit http://gerrit.cloudera.org:8080/settings Gerrit-Project: Impala-ASF Gerrit-Branch: master Gerrit-MessageType: comment Gerrit-Change-Id: I3416a79740eddc446c87f72bf1a85ed3f71af268 Gerrit-Change-Number: 21423 Gerrit-PatchSet: 7 Gerrit-Owner: Peter Rozsa <[email protected]> Gerrit-Reviewer: Daniel Becker <[email protected]> Gerrit-Reviewer: Gabor Kaszab <[email protected]> Gerrit-Reviewer: Impala Public Jenkins <[email protected]> Gerrit-Reviewer: Noemi Pap-Takacs <[email protected]> Gerrit-Reviewer: Peter Rozsa <[email protected]> Gerrit-Reviewer: Zoltan Borok-Nagy <[email protected]> Gerrit-Comment-Date: Thu, 13 Jun 2024 12:17:32 +0000 Gerrit-HasComments: Yes
