Lend your Support and Help For containerize installation

hetu-carbondata/src/main/java/io/hetu/core/plugin/carbondata/CarbondataInsertTableHandle.java

@@ -51,7 +51,7 @@ public CarbondataInsertTableHandle(@JsonProperty("schemaName") String schemaName
     {
         super(schemaName, tableName, inputColumns, pageSinkMetadata,
                 locationHandle, bucketProperty, tableStorageFormat,
-                partitionStorageFormat, isOverwrite);
+                partitionStorageFormat, isOverwrite, false);
 
         this.additionalConf = ImmutableMap.copyOf(requireNonNull(additionalConf, "additional conf map is null"));
     }

hetu-carbondata/src/main/java/io/hetu/core/plugin/carbondata/CarbondataMetadata.java

@@ -80,6 +80,7 @@
 import io.prestosql.spi.connector.ConnectorUpdateTableHandle;
 import io.prestosql.spi.connector.ConnectorVacuumTableHandle;
 import io.prestosql.spi.connector.ConnectorVacuumTableInfo;
+import io.prestosql.spi.connector.RetryMode;
 import io.prestosql.spi.connector.SchemaNotFoundException;
 import io.prestosql.spi.connector.SchemaTableName;
 import io.prestosql.spi.connector.TableAlreadyExistsException;
@@ -197,6 +198,7 @@
 import static io.prestosql.plugin.hive.HiveUtil.toPartitionValues;
 import static io.prestosql.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
 import static io.prestosql.spi.StandardErrorCode.NOT_SUPPORTED;
+import static io.prestosql.spi.connector.RetryMode.NO_RETRIES;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 import static java.util.stream.Collectors.joining;
@@ -345,6 +347,24 @@ public CarbondataInsertTableHandle beginInsert(ConnectorSession session, Connect
         return beginInsertUpdateInternal(session, tableHandle, false || isInsertExistingPartitionsOverwrite);
     }
 
+    @Override
+    public CarbondataInsertTableHandle beginInsert(ConnectorSession session, ConnectorTableHandle tableHandle, RetryMode retryMode)
+    {
+        if (retryMode != NO_RETRIES) {
+            throw new PrestoException(NOT_SUPPORTED, "Carbondata connector does not support insert with query retries enabled");
+        }
+        return beginInsert(session, tableHandle);
+    }
+
+    @Override
+    public CarbondataInsertTableHandle beginInsert(ConnectorSession session, ConnectorTableHandle tableHandle, boolean isOverwrite, RetryMode retryMode)
+    {
+        if (retryMode != NO_RETRIES) {
+            throw new PrestoException(NOT_SUPPORTED, "Carbondata connector does not support insert with query retries enabled");
+        }
+        return beginInsert(session, tableHandle, isOverwrite);
+    }
+
     @Override
     public CarbondataInsertTableHandle beginInsert(ConnectorSession session, ConnectorTableHandle tableHandle, boolean isOverwrite)
     {
@@ -410,6 +430,15 @@ public Optional<ConnectorOutputMetadata> finishInsert(ConnectorSession session,
         return connectorOutputMetadata;
     }
 
+    @Override
+    public CarbondataUpdateTableHandle beginUpdateAsInsert(ConnectorSession session, ConnectorTableHandle tableHandle, RetryMode retryMode)
+    {
+        if (retryMode != NO_RETRIES) {
+            throw new PrestoException(NOT_SUPPORTED, "Carbondata connector does not support update with query retries enabled");
+        }
+        return beginUpdateAsInsert(session, tableHandle);
+    }
+
     @Override
     public CarbondataUpdateTableHandle beginUpdateAsInsert(ConnectorSession session, ConnectorTableHandle tableHandle)
     {
@@ -463,6 +492,15 @@ public CarbondataUpdateTableHandle beginUpdateAsInsert(ConnectorSession session,
                 additionalConf);
     }
 
+    @Override
+    public CarbonDeleteAsInsertTableHandle beginDeletesAsInsert(ConnectorSession session, ConnectorTableHandle tableHandle, RetryMode retryMode)
+    {
+        if (retryMode != NO_RETRIES) {
+            throw new PrestoException(NOT_SUPPORTED, "Carbondata connector does not support delete with query retries enabled");
+        }
+        return beginDeletesAsInsert(session, tableHandle);
+    }
+
     @Override
     public CarbonDeleteAsInsertTableHandle beginDeletesAsInsert(ConnectorSession session, ConnectorTableHandle tableHandle)
     {
@@ -549,7 +587,7 @@ public Optional<ConnectorOutputMetadata> finishUpdateAsInsert(ConnectorSession s
                 updateTableHandle.getSchemaName(), updateTableHandle.getTableName(),
                 updateTableHandle.getInputColumns(), updateTableHandle.getPageSinkMetadata(),
                 updateTableHandle.getLocationHandle(), updateTableHandle.getBucketProperty(),
-                updateTableHandle.getTableStorageFormat(), updateTableHandle.getPartitionStorageFormat(), false);
+                updateTableHandle.getTableStorageFormat(), updateTableHandle.getPartitionStorageFormat(), false, updateTableHandle.isRetriesEnabled());
 
         autoCleanerTasks.add(new CarbondataAutoCleaner(table.get(), initialConfiguration, carbondataTableReader, metastore, hdfsEnvironment, user));
         return finishUpdateAndDelete(session, insertTableHandle, fragments, computedStatistics);
@@ -563,7 +601,7 @@ public Optional<ConnectorOutputMetadata> finishDeleteAsInsert(ConnectorSession s
                 deleteTableHandle.getSchemaName(), deleteTableHandle.getTableName(),
                 deleteTableHandle.getInputColumns(), deleteTableHandle.getPageSinkMetadata(),
                 deleteTableHandle.getLocationHandle(), deleteTableHandle.getBucketProperty(),
-                deleteTableHandle.getTableStorageFormat(), deleteTableHandle.getPartitionStorageFormat(), false);
+                deleteTableHandle.getTableStorageFormat(), deleteTableHandle.getPartitionStorageFormat(), false, deleteTableHandle.isRetriesEnabled());
 
         autoCleanerTasks.add(new CarbondataAutoCleaner(table.get(), initialConfiguration, carbondataTableReader, metastore, hdfsEnvironment, user));
         return finishUpdateAndDelete(session, insertTableHandle, fragments, computedStatistics);
@@ -1190,6 +1228,15 @@ public void getParametersForCreateTable(ConnectorSession session,
         tableProperties.putAll(getEmptyTableProperties(tableMetadata, bucketProperty, new HdfsEnvironment.HdfsContext(session, finalSchemaName, finalTableName)));
     }
 
+    @Override
+    public CarbondataOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Optional<ConnectorNewTableLayout> layout, RetryMode retryMode)
+    {
+        if (retryMode != NO_RETRIES) {
+            throw new PrestoException(NOT_SUPPORTED, "Carbondata connector does not support create table with query retries enabled");
+        }
+        return beginCreateTable(session, tableMetadata, layout);
+    }
+
     @Override
     public CarbondataOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Optional<ConnectorNewTableLayout> layout)
     {
@@ -1416,7 +1463,8 @@ protected void finishInsertOverwrite(ConnectorSession session, HiveInsertTableHa
                 partitionUpdate.getWritePath(),
                 partitionUpdate.getFileNames(),
                 partitionStatistics,
-                HiveACIDWriteType.INSERT_OVERWRITE);
+                HiveACIDWriteType.INSERT_OVERWRITE,
+                false);
         markSegmentsForDelete(session, table);
     }
 
@@ -1439,7 +1487,8 @@ protected void finishInsertInNewPartition(ConnectorSession session, HiveInsertTa
                     partitionUpdate.getWritePath(),
                     partitionUpdate.getFileNames(),
                     partitionStatistics,
-                    acidWriteType);
+                    acidWriteType,
+                    false);
         }
         else if (partitionUpdate.getUpdateMode() == PartitionUpdate.UpdateMode.APPEND) {
             Partition partition = buildPartitionObject(session, table, partitionUpdate);
@@ -1452,7 +1501,7 @@ else if (partitionUpdate.getUpdateMode() == PartitionUpdate.UpdateMode.APPEND) {
                     partitionUpdate.getStatistics(),
                     columnTypes,
                     getColumnStatistics(partitionComputedStatistics, partition.getValues()));
-            metastore.addPartition(session, handle.getSchemaName(), handle.getTableName(), partition, partitionUpdate.getWritePath(), partitionStatistics, acidWriteType);
+            metastore.addPartition(session, handle.getSchemaName(), handle.getTableName(), partition, partitionUpdate.getWritePath(), partitionStatistics, acidWriteType, Optional.empty(), false);
         }
     }
 

hetu-carbondata/src/main/java/io/hetu/core/plugin/carbondata/CarbondataOutputTableHandle.java

@@ -49,7 +49,7 @@ public CarbondataOutputTableHandle(@JsonProperty("schemaName") String schemaName
                                        @JsonProperty("additionalConf") Map<String, String> additionalConf)
     {
         super(schemaName, tableName, inputColumns, pageSinkMetadata, locationHandle, tableStorageFormat,
-                partitionStorageFormat, partitionedBy, bucketProperty, tableOwner, additionalTableParameters);
+                partitionStorageFormat, partitionedBy, bucketProperty, tableOwner, additionalTableParameters, false);
         this.additionalConf = ImmutableMap.copyOf(requireNonNull(additionalConf, "additional conf map is null"));
     }
 

presto-hive/src/main/java/io/prestosql/plugin/hive/HiveDeleteAsInsertTableHandle.java

@@ -46,6 +46,7 @@ public HiveDeleteAsInsertTableHandle(
                 bucketProperty,
                 tableStorageFormat,
                 partitionStorageFormat,
+                false,
                 false);
     }
 }

presto-hive/src/main/java/io/prestosql/plugin/hive/HiveInsertTableHandle.java

@@ -35,7 +35,8 @@ public HiveInsertTableHandle(
             @JsonProperty("bucketProperty") Optional<HiveBucketProperty> bucketProperty,
             @JsonProperty("tableStorageFormat") HiveStorageFormat tableStorageFormat,
             @JsonProperty("partitionStorageFormat") HiveStorageFormat partitionStorageFormat,
-            @JsonProperty("isOverwrite") boolean isOverwrite)
+            @JsonProperty("isOverwrite") boolean isOverwrite,
+            @JsonProperty("retriesEnabled") boolean retriesEnabled)
     {
         super(
                 schemaName,
@@ -46,6 +47,7 @@ public HiveInsertTableHandle(
                 bucketProperty,
                 tableStorageFormat,
                 partitionStorageFormat,
-                isOverwrite);
+                isOverwrite,
+                retriesEnabled);
     }
 }