[Iceberg]Accept and maintain autocommit context on starting transaction

Author: hantangwangd

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergAbstractMetadata.java

@@ -223,6 +223,7 @@ public abstract class IcebergAbstractMetadata
     protected Transaction transaction;
     protected final StatisticsFileCache statisticsFileCache;
     protected final IcebergTableProperties tableProperties;
+    protected final boolean autoCommitContext;
 
     private final StandardFunctionResolution functionResolution;
     private final ConcurrentMap<SchemaTableName, Table> icebergTables = new ConcurrentHashMap<>();
@@ -235,7 +236,8 @@ public IcebergAbstractMetadata(
             NodeVersion nodeVersion,
             FilterStatsCalculatorService filterStatsCalculatorService,
             StatisticsFileCache statisticsFileCache,
-            IcebergTableProperties tableProperties)
+            IcebergTableProperties tableProperties,
+            boolean autoCommitContext)
     {
         this.typeManager = requireNonNull(typeManager, "typeManager is null");
         this.commitTaskCodec = requireNonNull(commitTaskCodec, "commitTaskCodec is null");
@@ -245,6 +247,7 @@ public IcebergAbstractMetadata(
         this.filterStatsCalculatorService = requireNonNull(filterStatsCalculatorService, "filterStatsCalculatorService is null");
         this.statisticsFileCache = requireNonNull(statisticsFileCache, "statisticsFileCache is null");
         this.tableProperties = requireNonNull(tableProperties, "tableProperties is null");
+        this.autoCommitContext = autoCommitContext;
     }
 
     protected final Table getIcebergTable(ConnectorSession session, SchemaTableName schemaTableName)

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergConnector.java

@@ -184,12 +184,12 @@ public ConnectorAccessControl getAccessControl()
     }
 
     @Override
-    public ConnectorTransactionHandle beginTransaction(IsolationLevel isolationLevel, boolean readOnly)
+    public ConnectorTransactionHandle beginTransaction(IsolationLevel isolationLevel, boolean autoCommitContext, boolean readOnly)
     {
         checkConnectorSupports(SERIALIZABLE, isolationLevel);
         ConnectorTransactionHandle transaction = new HiveTransactionHandle();
         try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(getClass().getClassLoader())) {
-            transactionManager.put(transaction, metadataFactory.create());
+            transactionManager.put(transaction, metadataFactory.create(autoCommitContext));
         }
         return transaction;
     }

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergHiveMetadata.java

@@ -180,9 +180,11 @@ public IcebergHiveMetadata(
             StatisticsFileCache statisticsFileCache,
             ManifestFileCache manifestFileCache,
             IcebergTableProperties tableProperties,
-            ConnectorSystemConfig connectorSystemConfig)
+            ConnectorSystemConfig connectorSystemConfig,
+            boolean autoCommitContext)
     {
-        super(typeManager, functionResolution, rowExpressionService, commitTaskCodec, nodeVersion, filterStatsCalculatorService, statisticsFileCache, tableProperties);
+        super(typeManager, functionResolution, rowExpressionService, commitTaskCodec, nodeVersion,
+                filterStatsCalculatorService, statisticsFileCache, tableProperties, autoCommitContext);
         this.catalogName = requireNonNull(catalogName, "catalogName is null");
         this.metastore = requireNonNull(metastore, "metastore is null");
         this.hdfsEnvironment = requireNonNull(hdfsEnvironment, "hdfsEnvironment is null");

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergHiveMetadataFactory.java

@@ -81,6 +81,11 @@ public IcebergHiveMetadataFactory(
     }
 
     public ConnectorMetadata create()
+    {
+        return create(true);
+    }
+
+    public ConnectorMetadata create(boolean autoCommitContext)
     {
         return new IcebergHiveMetadata(
                 catalogName,
@@ -96,6 +101,7 @@ public ConnectorMetadata create()
                 statisticsFileCache,
                 manifestFileCache,
                 tableProperties,
-                connectorSystemConfig);
+                connectorSystemConfig,
+                autoCommitContext);
     }
 }

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergMetadataFactory.java

@@ -18,4 +18,6 @@
 public interface IcebergMetadataFactory
 {
     ConnectorMetadata create();
+
+    ConnectorMetadata create(boolean autoCommitContext);
 }

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergNativeMetadata.java

@@ -108,9 +108,11 @@ public IcebergNativeMetadata(
             NodeVersion nodeVersion,
             FilterStatsCalculatorService filterStatsCalculatorService,
             StatisticsFileCache statisticsFileCache,
-            IcebergTableProperties tableProperties)
+            IcebergTableProperties tableProperties,
+            boolean autoCommitContext)
     {
-        super(typeManager, functionResolution, rowExpressionService, commitTaskCodec, nodeVersion, filterStatsCalculatorService, statisticsFileCache, tableProperties);
+        super(typeManager, functionResolution, rowExpressionService, commitTaskCodec, nodeVersion,
+                filterStatsCalculatorService, statisticsFileCache, tableProperties, autoCommitContext);
         this.catalogFactory = requireNonNull(catalogFactory, "catalogFactory is null");
         this.catalogType = requireNonNull(catalogType, "catalogType is null");
         this.warehouseDataDir = Optional.ofNullable(catalogFactory.getCatalogWarehouseDataDir());

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergNativeMetadataFactory.java

@@ -68,6 +68,13 @@ public IcebergNativeMetadataFactory(
 
     public ConnectorMetadata create()
     {
-        return new IcebergNativeMetadata(catalogFactory, typeManager, functionResolution, rowExpressionService, commitTaskCodec, catalogType, nodeVersion, filterStatsCalculatorService, statisticsFileCache, tableProperties);
+        return create(true);
+    }
+
+    public ConnectorMetadata create(boolean autoCommitContext)
+    {
+        return new IcebergNativeMetadata(catalogFactory, typeManager, functionResolution,
+                rowExpressionService, commitTaskCodec, catalogType, nodeVersion,
+                filterStatsCalculatorService, statisticsFileCache, tableProperties, autoCommitContext);
     }
 }