IMPALA-11377: Handle concurrent Iceberg INSERT OVERWRITEs

Iceberg now supports validations in the ReplacePartitions API, an
initial snapshot id can be configured for the operation and Iceberg
will validate at commit if there were any conflicting change while the
ReplacePartitions was running.

This commit utilizes this new functionality, the Coordinator sends the
initial snapshot id to the Catalog, so the Catalog can validate the
operation.

Testing:
 - Added e2e tests.

Change-Id: I812d19736c8e563541b038970786de7710b59f31
Reviewed-on: http://gerrit.cloudera.org:8080/18648
Reviewed-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com>
Tested-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com>

Author: tmater

be/src/service/client-request-state.cc

@@ -1467,6 +1467,7 @@ Status ClientRequestState::UpdateCatalog() {
         ice_op.__set_iceberg_data_files_fb(
             dml_exec_state->CreateIcebergDataFilesVector());
         ice_op.__set_is_overwrite(finalize_params.is_overwrite);
+        ice_op.__set_initial_snapshot_id(finalize_params.initial_snapshot_id);
       }
 
       Status cnxn_status;

common/thrift/CatalogService.thrift

@@ -213,6 +213,9 @@ struct TIcebergOperationParam {
 
   // Is overwrite operation
   3: required bool is_overwrite = false;
+
+  // The snapshot id when the operation was started
+  4: optional i64 initial_snapshot_id;
 }
 
 // Per-partion info needed by Catalog to handle an INSERT.

common/thrift/Query.thrift

@@ -816,6 +816,9 @@ struct TFinalizeParams {
 
   // Stores the Iceberg spec id of the partition spec used for this INSERT.
   9: optional i32 spec_id;
+
+  // Stores the Iceberg snapshot id of the target table for INSERTs.
+  10: optional i64 initial_snapshot_id;
 }
 
 // Result of call to ImpalaPlanService/JniFrontend.CreateQueryRequest()

fe/src/main/java/org/apache/impala/service/Frontend.java

@@ -2259,6 +2259,7 @@ public static void addFinalizationParamsForInsert(
       } else if (targetTable instanceof FeIcebergTable) {
         FeIcebergTable iceTable = (FeIcebergTable)targetTable;
         finalizeParams.setSpec_id(iceTable.getDefaultPartitionSpecId());
+        finalizeParams.setInitial_snapshot_id(iceTable.snapshotId());
       } else {
         // TODO: Currently this flag only controls the removal of the query-level staging
         // directory. HdfsTableSink (that creates the staging dir) calculates the path

fe/src/main/java/org/apache/impala/service/IcebergCatalogOpExecutor.java

@@ -23,6 +23,7 @@
 import java.util.Map;
 
 import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.BaseReplacePartitions;
 import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
@@ -38,6 +39,7 @@
 import org.apache.iceberg.UpdateProperties;
 import org.apache.iceberg.UpdateSchema;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEventPropertyKey;
 import org.apache.impala.catalog.FeIcebergTable;
@@ -250,8 +252,10 @@ public void commit() {
 
   private static class DynamicOverwrite implements BatchWrite {
     final private ReplacePartitions replace;
-    public DynamicOverwrite(Transaction txn) {
+    final long initialSnapshotId;
+    public DynamicOverwrite(Transaction txn, long initialSnapshotId) {
       replace = txn.newReplacePartitions();
+      this.initialSnapshotId = initialSnapshotId;
     }
 
     @Override
@@ -261,6 +265,9 @@ public void addFile(DataFile file) {
 
     @Override
     public void commit() {
+      replace.validateFromSnapshot(initialSnapshotId);
+      replace.validateNoConflictingData();
+      replace.validateNoConflictingDeletes();
       replace.commit();
     }
   }
@@ -275,7 +282,7 @@ public static void appendFiles(FeIcebergTable feIcebergTable, Transaction txn,
     List<ByteBuffer> dataFilesFb = icebergOp.getIceberg_data_files_fb();
     BatchWrite batchWrite;
     if (icebergOp.isIs_overwrite()) {
-      batchWrite = new DynamicOverwrite(txn);
+      batchWrite = new DynamicOverwrite(txn, icebergOp.getInitial_snapshot_id());
     } else {
       batchWrite = new Append(txn);
     }
@@ -297,7 +304,11 @@ public static void appendFiles(FeIcebergTable feIcebergTable, Transaction txn,
       if (partitionData != null) builder.withPartition(partitionData);
       batchWrite.addFile(builder.build());
     }
-    batchWrite.commit();
+    try {
+      batchWrite.commit();
+    } catch (ValidationException e) {
+      throw new ImpalaRuntimeException(e.getMessage(), e);
+    }
   }
 
   private static Metrics buildDataFileMetrics(FbIcebergDataFile dataFile) {

tests/query_test/test_iceberg.py

@@ -28,6 +28,7 @@
 from avro.io import DatumReader
 import json
 
+from tests.beeswax.impala_beeswax import ImpalaBeeswaxException
 from tests.common.impala_test_suite import ImpalaTestSuite, LOG
 from tests.common.skip import SkipIf
 
@@ -71,18 +72,54 @@ def test_drop_incomplete_table(self, vector, unique_database):
     self.hdfs_client.delete_file_dir(cat_location, True)
     self.execute_query_expect_success(self.client, """drop table {0}""".format(tbl_name))
 
-  @SkipIf.not_hdfs
   def test_insert(self, vector, unique_database):
     self.run_test_case('QueryTest/iceberg-insert', vector, use_db=unique_database)
 
   def test_partitioned_insert(self, vector, unique_database):
     self.run_test_case('QueryTest/iceberg-partitioned-insert', vector,
         use_db=unique_database)
 
-  @SkipIf.not_hdfs
   def test_insert_overwrite(self, vector, unique_database):
+    """Run iceberg-overwrite tests, then test that INSERT INTO/OVERWRITE queries running
+    concurrently with a long running INSERT OVERWRITE are handled gracefully. query_a is
+    started before query_b/query_c, but query_b/query_c supposed to finish before query_a.
+    query_a should fail because the overwrite should not erase query_b/query_c's result.
+    """
+    # Run iceberg-overwrite.test
     self.run_test_case('QueryTest/iceberg-overwrite', vector, use_db=unique_database)
 
+    # Create test dataset for concurrency tests and warm-up the test table
+    tbl_name = unique_database + ".overwrite_tbl"
+    self.client.execute("""create table {0} (i int)
+        partitioned by spec (truncate(3, i))
+        stored as iceberg""".format(tbl_name))
+    self.client.execute("insert into {0} values (1), (2), (3);".format(tbl_name))
+
+    # Test queries: 'a' is the long running query while 'b' and 'c' are the short ones
+    query_a = """insert overwrite {0} select sleep(5000);""".format(tbl_name)
+    query_b = """insert overwrite {0} select * from {0};""".format(tbl_name)
+    query_c = """insert into {0} select * from {0};""".format(tbl_name)
+
+    # Test concurrent INSERT OVERWRITEs, the exception closes the query handle.
+    handle = self.client.execute_async(query_a)
+    time.sleep(1)
+    self.client.execute(query_b)
+    try:
+      self.client.wait_for_finished_timeout(handle, 30)
+      assert False
+    except ImpalaBeeswaxException as e:
+      assert "Found conflicting files" in str(e)
+
+    # Test INSERT INTO during INSERT OVERWRITE, the exception closes the query handle.
+    handle = self.client.execute_async(query_a)
+    time.sleep(1)
+    self.client.execute(query_c)
+    try:
+      self.client.wait_for_finished_timeout(handle, 30)
+      assert False
+    except ImpalaBeeswaxException as e:
+      assert "Found conflicting files" in str(e)
+
   def test_ctas(self, vector, unique_database):
     self.run_test_case('QueryTest/iceberg-ctas', vector, use_db=unique_database)