You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@xtable.apache.org by "sudhar91 (via GitHub)" <gi...@apache.org> on 2024/12/07 07:38:38 UTC
[PR] Sync parquet to other file formats [incubator-xtable]
sudhar91 opened a new pull request, #592:
URL: https://github.com/apache/incubator-xtable/pull/592
Issue : https://github.com/apache/incubator-xtable/issues/553
## What is the purpose of the pull request
*This PR is for changes made to sync parquet to delta ,hudi & Iceberg*
## Brief change log
- *Changes to sync from parquet to delta ,hudi & iceberg*
- *Handle incremental sync*
Co Authored by : @sundarshankar89
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sync parquet to other file formats [incubator-xtable]
Posted by "sudhar91 (via GitHub)" <gi...@apache.org>.
sudhar91 commented on code in PR #592:
URL: https://github.com/apache/incubator-xtable/pull/592#discussion_r1885714144
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetPartitionHelper.java:
##########
@@ -0,0 +1,69 @@
+package org.apache.xtable.parquet;
+
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.PartitionTransformType;
+import org.apache.xtable.model.stat.PartitionValue;
+import org.apache.xtable.model.stat.Range;
+import org.apache.xtable.schema.SchemaFieldFinder;
+
+public class ParquetPartitionHelper {
+ private static final ParquetPartitionHelper INSTANCE = new ParquetPartitionHelper();
+
+ public static ParquetPartitionHelper getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalPartitionField> getInternalPartitionField(
+ Set<String> partitionList, InternalSchema schema) {
+ List<InternalPartitionField> partitionFields = new ArrayList<>();
+
+ for (String partitionKey : partitionList) {
+
+ partitionFields.add(
+ InternalPartitionField.builder()
+ .sourceField(SchemaFieldFinder.getInstance().findFieldByPath(schema, partitionKey))
+ .transformType(PartitionTransformType.VALUE)
+ .build());
+ }
+
+ return partitionFields;
+ }
+
+ // TODO logic is too complicated can be simplified
+ public List<PartitionValue> getPartitionValue(
+ String basePath,
+ String filePath,
+ InternalSchema schema,
+ Map<String, List<String>> partitionInfo) {
+ List<PartitionValue> partitionValues = new ArrayList<>();
+ java.nio.file.Path base = Paths.get(basePath).normalize();
Review Comment:
Noted , overall i need to simplify the logic as it is bit complicated
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sync parquet to other file formats [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #592:
URL: https://github.com/apache/incubator-xtable/pull/592#discussion_r1875038720
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java:
##########
@@ -0,0 +1,209 @@
+package org.apache.xtable.parquet;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.*;
+import java.util.stream.Collectors;
+import lombok.Builder;
+import lombok.NonNull;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.xtable.avro.AvroSchemaConverter;
+import org.apache.xtable.model.*;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.*;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Builder
+public class ParquetConversionSource implements ConversionSource<Long> {
+
+ private final String tableName;
+ private final String basePath;
+ @NonNull private final Configuration hadoopConf;
+
+ @Builder.Default
+ private static final AvroSchemaConverter schemaExtractor = AvroSchemaConverter.getInstance();
+
+ @Builder.Default private static final FileSystemHelper fsHelper = FileSystemHelper.getInstance();
+
+ @Builder.Default
+ private static final ParquetMetadataExtractor parquetMetadataExtractor =
+ ParquetMetadataExtractor.getInstance();
+
+ @Builder.Default
+ private static final ParquetPartitionHelper parquetPartitionHelper =
+ ParquetPartitionHelper.getInstance();
+
+ private Map<String, List<String>> initPartitionInfo() {
+ return fsHelper.getPartitionFromDirectoryStructure(
+ hadoopConf, basePath, Collections.emptyMap());
+ }
+
+ /**
+ * To infer schema getting the latest file assumption is that latest file will have new fields
+ *
+ * @param modificationTime the commit to consider for reading the table state
+ * @return
+ */
+ @Override
+ public InternalTable getTable(Long modificationTime) {
+
+ Optional<LocatedFileStatus> latestFile =
+ fsHelper
+ .getParquetFiles(hadoopConf, basePath)
+ .max(Comparator.comparing(FileStatus::getModificationTime));
+
+ ParquetMetadata parquetMetadata =
+ parquetMetadataExtractor.readParquetMetadata(hadoopConf, latestFile.get().getPath());
+ Schema tableSchema =
+ new org.apache.parquet.avro.AvroSchemaConverter()
+ .convert(parquetMetadata.getFileMetaData().getSchema());
+
+ Set<String> partitionKeys = initPartitionInfo().keySet();
+
+ // merge schema of partition into original as partition is not part of parquet fie
+ if (!partitionKeys.isEmpty()) {
+ tableSchema = mergeAvroSchema(tableSchema, partitionKeys);
+ }
+ InternalSchema schema = schemaExtractor.toInternalSchema(tableSchema);
+
+ List<InternalPartitionField> partitionFields =
+ partitionKeys.isEmpty()
+ ? Collections.emptyList()
+ : parquetPartitionHelper.getInternalPartitionField(partitionKeys, schema);
+ DataLayoutStrategy dataLayoutStrategy =
+ partitionFields.isEmpty()
+ ? DataLayoutStrategy.FLAT
+ : DataLayoutStrategy.HIVE_STYLE_PARTITION;
+ return InternalTable.builder()
+ .tableFormat(TableFormat.PARQUET)
+ .basePath(basePath)
+ .name(tableName)
+ .layoutStrategy(dataLayoutStrategy)
+ .partitioningFields(partitionFields)
+ .readSchema(schema)
+ .latestCommitTime(Instant.ofEpochMilli(latestFile.get().getModificationTime()))
+ .build();
+ }
+
+ /**
+ * Here to get current snapshot listing all files hence the -1 is being passed
+ *
+ * @return
+ */
+ @Override
+ public InternalSnapshot getCurrentSnapshot() {
+
+ List<LocatedFileStatus> latestFile =
+ fsHelper.getParquetFiles(hadoopConf, basePath).collect(Collectors.toList());
+ Map<String, List<String>> partitionInfo = initPartitionInfo();
+ InternalTable table = getTable(-1L);
+ List<InternalDataFile> internalDataFiles =
+ latestFile.stream()
+ .map(
+ file ->
+ InternalDataFile.builder()
+ .physicalPath(file.getPath().toString())
+ .fileFormat(FileFormat.APACHE_PARQUET)
+ .fileSizeBytes(file.getLen())
+ .partitionValues(
+ parquetPartitionHelper.getPartitionValue(
+ basePath,
+ file.getPath().toString(),
+ table.getReadSchema(),
+ partitionInfo))
+ .lastModified(file.getModificationTime())
+ .columnStats(
+ parquetMetadataExtractor.getColumnStatsForaFile(
+ hadoopConf, file, table))
+ .build())
+ .collect(Collectors.toList());
+
+ return InternalSnapshot.builder()
+ .table(table)
+ .partitionedDataFiles(PartitionFileGroup.fromFiles(internalDataFiles))
+ .build();
+ }
+
+ /**
+ * Whenever new file is added , condition to get new file is listing files whose modification time
+ * is greater than previous ysnc
+ *
+ * @param modificationTime commit to capture table changes for.
+ * @return
+ */
+ @Override
+ public TableChange getTableChangeForCommit(Long modificationTime) {
+ List<FileStatus> tableChanges =
+ fsHelper
+ .getParquetFiles(hadoopConf, basePath)
+ .filter(fileStatus -> fileStatus.getModificationTime() > modificationTime)
+ .collect(Collectors.toList());
+ // TODO avoid doing full list of directory to get schema , just argument of modification time
+ // needs to be tweaked
+ InternalTable internalTable = getTable(-1L);
+ Set<InternalDataFile> internalDataFiles = new HashSet<>();
+ Map<String, List<String>> partitionInfo = initPartitionInfo();
+ for (FileStatus tableStatus : tableChanges) {
+ internalDataFiles.add(
+ InternalDataFile.builder()
Review Comment:
We also need `recordCount` which we should be able to get from the column stats
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sync parquet to other file formats [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #592:
URL: https://github.com/apache/incubator-xtable/pull/592#discussion_r1875033692
##########
xtable-api/src/main/java/org/apache/xtable/model/storage/TableFormat.java:
##########
@@ -27,8 +27,9 @@ public class TableFormat {
public static final String HUDI = "HUDI";
public static final String ICEBERG = "ICEBERG";
public static final String DELTA = "DELTA";
+ public static final String PARQUET="PARQUET";
Review Comment:
Please run `mvn spotless:apply` to clean up some of the formatting issues in the draft
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java:
##########
@@ -0,0 +1,209 @@
+package org.apache.xtable.parquet;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.*;
+import java.util.stream.Collectors;
+import lombok.Builder;
+import lombok.NonNull;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.xtable.avro.AvroSchemaConverter;
+import org.apache.xtable.model.*;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.*;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Builder
+public class ParquetConversionSource implements ConversionSource<Long> {
+
+ private final String tableName;
+ private final String basePath;
+ @NonNull private final Configuration hadoopConf;
+
+ @Builder.Default
+ private static final AvroSchemaConverter schemaExtractor = AvroSchemaConverter.getInstance();
+
+ @Builder.Default private static final FileSystemHelper fsHelper = FileSystemHelper.getInstance();
+
+ @Builder.Default
+ private static final ParquetMetadataExtractor parquetMetadataExtractor =
Review Comment:
Is there an implementation for this class missing?
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSourceProvider.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.parquet;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+
+/** A concrete implementation of {@link ConversionSourceProvider} for Delta Lake table format. */
+public class ParquetConversionSourceProvider extends ConversionSourceProvider<Long> {
+ @Override
+ public ParquetConversionSource getConversionSourceInstance(SourceTable sourceTable) {
+
+return ParquetConversionSource.builder()
+ .tableName(sourceTable.getName())
+ .basePath(sourceTable.getBasePath())
+ .hadoopConf(new Configuration())
Review Comment:
there is an `init` method called with the hadoop configuration, you should be able to simply use `hadoopConf` here
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetPartitionHelper.java:
##########
@@ -0,0 +1,69 @@
+package org.apache.xtable.parquet;
+
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.PartitionTransformType;
+import org.apache.xtable.model.stat.PartitionValue;
+import org.apache.xtable.model.stat.Range;
+import org.apache.xtable.schema.SchemaFieldFinder;
+
+public class ParquetPartitionHelper {
Review Comment:
nitpick on naming, "helper" is a bit vague and ends up being a place where developers will dump all sorts of functionality. This is a `ParquetPartitionExtractor` based on the naming conventions we've used in other formats
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java:
##########
@@ -0,0 +1,209 @@
+package org.apache.xtable.parquet;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.*;
+import java.util.stream.Collectors;
+import lombok.Builder;
+import lombok.NonNull;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.xtable.avro.AvroSchemaConverter;
+import org.apache.xtable.model.*;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.*;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Builder
+public class ParquetConversionSource implements ConversionSource<Long> {
+
+ private final String tableName;
+ private final String basePath;
+ @NonNull private final Configuration hadoopConf;
+
+ @Builder.Default
+ private static final AvroSchemaConverter schemaExtractor = AvroSchemaConverter.getInstance();
+
+ @Builder.Default private static final FileSystemHelper fsHelper = FileSystemHelper.getInstance();
+
+ @Builder.Default
+ private static final ParquetMetadataExtractor parquetMetadataExtractor =
+ ParquetMetadataExtractor.getInstance();
+
+ @Builder.Default
+ private static final ParquetPartitionHelper parquetPartitionHelper =
+ ParquetPartitionHelper.getInstance();
+
+ private Map<String, List<String>> initPartitionInfo() {
+ return fsHelper.getPartitionFromDirectoryStructure(
+ hadoopConf, basePath, Collections.emptyMap());
+ }
+
+ /**
+ * To infer schema getting the latest file assumption is that latest file will have new fields
+ *
+ * @param modificationTime the commit to consider for reading the table state
+ * @return
+ */
+ @Override
+ public InternalTable getTable(Long modificationTime) {
+
+ Optional<LocatedFileStatus> latestFile =
+ fsHelper
+ .getParquetFiles(hadoopConf, basePath)
+ .max(Comparator.comparing(FileStatus::getModificationTime));
Review Comment:
Is there a way to push down this filter so we don't need to iterate through all files under the base path? Maybe we can even limit the file listing to return files created after the modificationTime?
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetPartitionHelper.java:
##########
@@ -0,0 +1,69 @@
+package org.apache.xtable.parquet;
+
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.PartitionTransformType;
+import org.apache.xtable.model.stat.PartitionValue;
+import org.apache.xtable.model.stat.Range;
+import org.apache.xtable.schema.SchemaFieldFinder;
+
+public class ParquetPartitionHelper {
+ private static final ParquetPartitionHelper INSTANCE = new ParquetPartitionHelper();
+
+ public static ParquetPartitionHelper getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalPartitionField> getInternalPartitionField(
+ Set<String> partitionList, InternalSchema schema) {
+ List<InternalPartitionField> partitionFields = new ArrayList<>();
+
+ for (String partitionKey : partitionList) {
+
+ partitionFields.add(
+ InternalPartitionField.builder()
+ .sourceField(SchemaFieldFinder.getInstance().findFieldByPath(schema, partitionKey))
+ .transformType(PartitionTransformType.VALUE)
+ .build());
+ }
+
+ return partitionFields;
+ }
+
+ // TODO logic is too complicated can be simplified
+ public List<PartitionValue> getPartitionValue(
+ String basePath,
+ String filePath,
+ InternalSchema schema,
+ Map<String, List<String>> partitionInfo) {
+ List<PartitionValue> partitionValues = new ArrayList<>();
+ java.nio.file.Path base = Paths.get(basePath).normalize();
+ java.nio.file.Path file = Paths.get(filePath).normalize();
+ java.nio.file.Path relative = base.relativize(file);
+ for (Map.Entry<String, List<String>> entry : partitionInfo.entrySet()) {
+ String key = entry.getKey();
+ List<String> values = entry.getValue();
+ for (String value : values) {
+ String pathCheck = key + "=" + value;
+ if (relative.startsWith(pathCheck)) {
+ System.out.println("Relative " + relative + " " + pathCheck);
Review Comment:
Be sure to remove this println
##########
xtable-core/src/main/java/org/apache/xtable/parquet/FileSystemHelper.java:
##########
@@ -0,0 +1,81 @@
+package org.apache.xtable.parquet;
+
+import java.io.IOException;
+import java.util.*;
Review Comment:
nitpick: we're avoiding the use of `*` imports in this repo
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java:
##########
@@ -0,0 +1,209 @@
+package org.apache.xtable.parquet;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.*;
+import java.util.stream.Collectors;
+import lombok.Builder;
+import lombok.NonNull;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.xtable.avro.AvroSchemaConverter;
+import org.apache.xtable.model.*;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.*;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Builder
+public class ParquetConversionSource implements ConversionSource<Long> {
+
+ private final String tableName;
+ private final String basePath;
+ @NonNull private final Configuration hadoopConf;
+
+ @Builder.Default
+ private static final AvroSchemaConverter schemaExtractor = AvroSchemaConverter.getInstance();
+
+ @Builder.Default private static final FileSystemHelper fsHelper = FileSystemHelper.getInstance();
+
+ @Builder.Default
+ private static final ParquetMetadataExtractor parquetMetadataExtractor =
+ ParquetMetadataExtractor.getInstance();
+
+ @Builder.Default
+ private static final ParquetPartitionHelper parquetPartitionHelper =
+ ParquetPartitionHelper.getInstance();
+
+ private Map<String, List<String>> initPartitionInfo() {
+ return fsHelper.getPartitionFromDirectoryStructure(
+ hadoopConf, basePath, Collections.emptyMap());
+ }
+
+ /**
+ * To infer schema getting the latest file assumption is that latest file will have new fields
+ *
+ * @param modificationTime the commit to consider for reading the table state
+ * @return
+ */
+ @Override
+ public InternalTable getTable(Long modificationTime) {
+
+ Optional<LocatedFileStatus> latestFile =
+ fsHelper
+ .getParquetFiles(hadoopConf, basePath)
+ .max(Comparator.comparing(FileStatus::getModificationTime));
+
+ ParquetMetadata parquetMetadata =
+ parquetMetadataExtractor.readParquetMetadata(hadoopConf, latestFile.get().getPath());
+ Schema tableSchema =
+ new org.apache.parquet.avro.AvroSchemaConverter()
+ .convert(parquetMetadata.getFileMetaData().getSchema());
+
+ Set<String> partitionKeys = initPartitionInfo().keySet();
+
+ // merge schema of partition into original as partition is not part of parquet fie
+ if (!partitionKeys.isEmpty()) {
+ tableSchema = mergeAvroSchema(tableSchema, partitionKeys);
+ }
+ InternalSchema schema = schemaExtractor.toInternalSchema(tableSchema);
+
+ List<InternalPartitionField> partitionFields =
+ partitionKeys.isEmpty()
+ ? Collections.emptyList()
+ : parquetPartitionHelper.getInternalPartitionField(partitionKeys, schema);
+ DataLayoutStrategy dataLayoutStrategy =
+ partitionFields.isEmpty()
+ ? DataLayoutStrategy.FLAT
+ : DataLayoutStrategy.HIVE_STYLE_PARTITION;
+ return InternalTable.builder()
+ .tableFormat(TableFormat.PARQUET)
+ .basePath(basePath)
+ .name(tableName)
+ .layoutStrategy(dataLayoutStrategy)
+ .partitioningFields(partitionFields)
+ .readSchema(schema)
+ .latestCommitTime(Instant.ofEpochMilli(latestFile.get().getModificationTime()))
+ .build();
+ }
+
+ /**
+ * Here to get current snapshot listing all files hence the -1 is being passed
+ *
+ * @return
+ */
+ @Override
+ public InternalSnapshot getCurrentSnapshot() {
+
+ List<LocatedFileStatus> latestFile =
+ fsHelper.getParquetFiles(hadoopConf, basePath).collect(Collectors.toList());
+ Map<String, List<String>> partitionInfo = initPartitionInfo();
+ InternalTable table = getTable(-1L);
+ List<InternalDataFile> internalDataFiles =
+ latestFile.stream()
+ .map(
+ file ->
+ InternalDataFile.builder()
+ .physicalPath(file.getPath().toString())
+ .fileFormat(FileFormat.APACHE_PARQUET)
+ .fileSizeBytes(file.getLen())
+ .partitionValues(
+ parquetPartitionHelper.getPartitionValue(
+ basePath,
+ file.getPath().toString(),
+ table.getReadSchema(),
+ partitionInfo))
+ .lastModified(file.getModificationTime())
+ .columnStats(
+ parquetMetadataExtractor.getColumnStatsForaFile(
+ hadoopConf, file, table))
+ .build())
+ .collect(Collectors.toList());
+
+ return InternalSnapshot.builder()
+ .table(table)
+ .partitionedDataFiles(PartitionFileGroup.fromFiles(internalDataFiles))
+ .build();
+ }
+
+ /**
+ * Whenever new file is added , condition to get new file is listing files whose modification time
+ * is greater than previous ysnc
+ *
+ * @param modificationTime commit to capture table changes for.
+ * @return
+ */
+ @Override
+ public TableChange getTableChangeForCommit(Long modificationTime) {
+ List<FileStatus> tableChanges =
+ fsHelper
+ .getParquetFiles(hadoopConf, basePath)
+ .filter(fileStatus -> fileStatus.getModificationTime() > modificationTime)
+ .collect(Collectors.toList());
+ // TODO avoid doing full list of directory to get schema , just argument of modification time
+ // needs to be tweaked
+ InternalTable internalTable = getTable(-1L);
+ Set<InternalDataFile> internalDataFiles = new HashSet<>();
+ Map<String, List<String>> partitionInfo = initPartitionInfo();
+ for (FileStatus tableStatus : tableChanges) {
+ internalDataFiles.add(
+ InternalDataFile.builder()
+ .physicalPath(tableStatus.getPath().toString())
+ .partitionValues(
+ parquetPartitionHelper.getPartitionValue(
+ basePath,
+ tableStatus.getPath().toString(),
+ internalTable.getReadSchema(),
+ partitionInfo))
+ .lastModified(tableStatus.getModificationTime())
+ .fileSizeBytes(tableStatus.getLen())
+ .columnStats(
+ parquetMetadataExtractor.getColumnStatsForaFile(
+ hadoopConf, tableStatus, internalTable))
+ .build());
+ }
+
+ return TableChange.builder()
+ .tableAsOfChange(internalTable)
+ .filesDiff(DataFilesDiff.builder().filesAdded(internalDataFiles).build())
+ .build();
+ }
+
+ @Override
+ public CommitsBacklog<Long> getCommitsBacklog(
+ InstantsForIncrementalSync instantsForIncrementalSync) {
+
+ List<Long> commitsToProcess =
+ Collections.singletonList(instantsForIncrementalSync.getLastSyncInstant().toEpochMilli());
+
+ return CommitsBacklog.<Long>builder().commitsToProcess(commitsToProcess).build();
+ }
+
+ // TODO Need to understnad how this needs to be implemented should _SUCCESS or .staging dir needs
+ // to be checked
+ @Override
+ public boolean isIncrementalSyncSafeFrom(Instant instant) {
+ return true;
+ }
+
+ @Override
+ public void close() throws IOException {}
+
+ private Schema mergeAvroSchema(Schema internalSchema, Set<String> parititonFields) {
+
+ SchemaBuilder.FieldAssembler<Schema> fieldAssembler =
+ SchemaBuilder.record(internalSchema.getName()).fields();
+ for (Schema.Field field : internalSchema.getFields()) {
+ fieldAssembler = fieldAssembler.name(field.name()).type(field.schema()).noDefault();
Review Comment:
Can the internal schema have defaults? Can it also have docs on fields? those would be dropped with this code
##########
xtable-core/src/main/java/org/apache/xtable/parquet/FileSystemHelper.java:
##########
@@ -0,0 +1,81 @@
+package org.apache.xtable.parquet;
+
+import java.io.IOException;
+import java.util.*;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+
+public class FileSystemHelper {
Review Comment:
I think it would be a good idea to define an interface for getting the parquet files for the table and changes since the last run. Right now this is all being done through file listing but we should consider a case where someone implements a way to poll the changes through s3 events.
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java:
##########
@@ -0,0 +1,209 @@
+package org.apache.xtable.parquet;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.*;
+import java.util.stream.Collectors;
+import lombok.Builder;
+import lombok.NonNull;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.xtable.avro.AvroSchemaConverter;
+import org.apache.xtable.model.*;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.*;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Builder
+public class ParquetConversionSource implements ConversionSource<Long> {
+
+ private final String tableName;
+ private final String basePath;
+ @NonNull private final Configuration hadoopConf;
+
+ @Builder.Default
+ private static final AvroSchemaConverter schemaExtractor = AvroSchemaConverter.getInstance();
+
+ @Builder.Default private static final FileSystemHelper fsHelper = FileSystemHelper.getInstance();
+
+ @Builder.Default
+ private static final ParquetMetadataExtractor parquetMetadataExtractor =
+ ParquetMetadataExtractor.getInstance();
+
+ @Builder.Default
+ private static final ParquetPartitionHelper parquetPartitionHelper =
+ ParquetPartitionHelper.getInstance();
+
+ private Map<String, List<String>> initPartitionInfo() {
+ return fsHelper.getPartitionFromDirectoryStructure(
+ hadoopConf, basePath, Collections.emptyMap());
+ }
+
+ /**
+ * To infer schema getting the latest file assumption is that latest file will have new fields
+ *
+ * @param modificationTime the commit to consider for reading the table state
+ * @return
+ */
+ @Override
+ public InternalTable getTable(Long modificationTime) {
+
+ Optional<LocatedFileStatus> latestFile =
+ fsHelper
+ .getParquetFiles(hadoopConf, basePath)
+ .max(Comparator.comparing(FileStatus::getModificationTime));
+
+ ParquetMetadata parquetMetadata =
+ parquetMetadataExtractor.readParquetMetadata(hadoopConf, latestFile.get().getPath());
+ Schema tableSchema =
+ new org.apache.parquet.avro.AvroSchemaConverter()
+ .convert(parquetMetadata.getFileMetaData().getSchema());
+
+ Set<String> partitionKeys = initPartitionInfo().keySet();
+
+ // merge schema of partition into original as partition is not part of parquet fie
+ if (!partitionKeys.isEmpty()) {
+ tableSchema = mergeAvroSchema(tableSchema, partitionKeys);
+ }
+ InternalSchema schema = schemaExtractor.toInternalSchema(tableSchema);
+
+ List<InternalPartitionField> partitionFields =
+ partitionKeys.isEmpty()
+ ? Collections.emptyList()
+ : parquetPartitionHelper.getInternalPartitionField(partitionKeys, schema);
+ DataLayoutStrategy dataLayoutStrategy =
+ partitionFields.isEmpty()
+ ? DataLayoutStrategy.FLAT
+ : DataLayoutStrategy.HIVE_STYLE_PARTITION;
+ return InternalTable.builder()
+ .tableFormat(TableFormat.PARQUET)
+ .basePath(basePath)
+ .name(tableName)
+ .layoutStrategy(dataLayoutStrategy)
+ .partitioningFields(partitionFields)
+ .readSchema(schema)
+ .latestCommitTime(Instant.ofEpochMilli(latestFile.get().getModificationTime()))
+ .build();
+ }
+
+ /**
+ * Here to get current snapshot listing all files hence the -1 is being passed
+ *
+ * @return
+ */
+ @Override
+ public InternalSnapshot getCurrentSnapshot() {
+
+ List<LocatedFileStatus> latestFile =
+ fsHelper.getParquetFiles(hadoopConf, basePath).collect(Collectors.toList());
+ Map<String, List<String>> partitionInfo = initPartitionInfo();
+ InternalTable table = getTable(-1L);
+ List<InternalDataFile> internalDataFiles =
+ latestFile.stream()
+ .map(
+ file ->
+ InternalDataFile.builder()
+ .physicalPath(file.getPath().toString())
+ .fileFormat(FileFormat.APACHE_PARQUET)
+ .fileSizeBytes(file.getLen())
+ .partitionValues(
+ parquetPartitionHelper.getPartitionValue(
+ basePath,
+ file.getPath().toString(),
+ table.getReadSchema(),
+ partitionInfo))
+ .lastModified(file.getModificationTime())
+ .columnStats(
+ parquetMetadataExtractor.getColumnStatsForaFile(
+ hadoopConf, file, table))
+ .build())
+ .collect(Collectors.toList());
+
+ return InternalSnapshot.builder()
+ .table(table)
+ .partitionedDataFiles(PartitionFileGroup.fromFiles(internalDataFiles))
+ .build();
+ }
+
+ /**
+ * Whenever new file is added , condition to get new file is listing files whose modification time
+ * is greater than previous ysnc
+ *
+ * @param modificationTime commit to capture table changes for.
+ * @return
+ */
+ @Override
+ public TableChange getTableChangeForCommit(Long modificationTime) {
+ List<FileStatus> tableChanges =
+ fsHelper
+ .getParquetFiles(hadoopConf, basePath)
+ .filter(fileStatus -> fileStatus.getModificationTime() > modificationTime)
+ .collect(Collectors.toList());
+ // TODO avoid doing full list of directory to get schema , just argument of modification time
+ // needs to be tweaked
+ InternalTable internalTable = getTable(-1L);
Review Comment:
This call will also scan the file system if I'm understanding this correctly, can we avoid that? It can be expensive for large tables
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetPartitionHelper.java:
##########
@@ -0,0 +1,69 @@
+package org.apache.xtable.parquet;
+
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.PartitionTransformType;
+import org.apache.xtable.model.stat.PartitionValue;
+import org.apache.xtable.model.stat.Range;
+import org.apache.xtable.schema.SchemaFieldFinder;
+
+public class ParquetPartitionHelper {
+ private static final ParquetPartitionHelper INSTANCE = new ParquetPartitionHelper();
+
+ public static ParquetPartitionHelper getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalPartitionField> getInternalPartitionField(
+ Set<String> partitionList, InternalSchema schema) {
+ List<InternalPartitionField> partitionFields = new ArrayList<>();
+
+ for (String partitionKey : partitionList) {
+
+ partitionFields.add(
+ InternalPartitionField.builder()
+ .sourceField(SchemaFieldFinder.getInstance().findFieldByPath(schema, partitionKey))
+ .transformType(PartitionTransformType.VALUE)
+ .build());
+ }
+
+ return partitionFields;
+ }
+
+ // TODO logic is too complicated can be simplified
+ public List<PartitionValue> getPartitionValue(
+ String basePath,
+ String filePath,
+ InternalSchema schema,
+ Map<String, List<String>> partitionInfo) {
+ List<PartitionValue> partitionValues = new ArrayList<>();
+ java.nio.file.Path base = Paths.get(basePath).normalize();
Review Comment:
If you're always going to convert the basePath, you should try to find a way to convert it once in the caller and pass it in.
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java:
##########
@@ -0,0 +1,209 @@
+package org.apache.xtable.parquet;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.*;
+import java.util.stream.Collectors;
+import lombok.Builder;
+import lombok.NonNull;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.xtable.avro.AvroSchemaConverter;
+import org.apache.xtable.model.*;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.*;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Builder
+public class ParquetConversionSource implements ConversionSource<Long> {
Review Comment:
I think it may be a bit more robust if we use a time interval instead of a single `long` here. Then you will be able to draw a clear boundary for each run of the conversion source, what are you thoughts?
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java:
##########
@@ -0,0 +1,209 @@
+package org.apache.xtable.parquet;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.*;
+import java.util.stream.Collectors;
+import lombok.Builder;
+import lombok.NonNull;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.xtable.avro.AvroSchemaConverter;
+import org.apache.xtable.model.*;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.*;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Builder
+public class ParquetConversionSource implements ConversionSource<Long> {
+
+ private final String tableName;
+ private final String basePath;
+ @NonNull private final Configuration hadoopConf;
+
+ @Builder.Default
+ private static final AvroSchemaConverter schemaExtractor = AvroSchemaConverter.getInstance();
+
+ @Builder.Default private static final FileSystemHelper fsHelper = FileSystemHelper.getInstance();
+
+ @Builder.Default
+ private static final ParquetMetadataExtractor parquetMetadataExtractor =
+ ParquetMetadataExtractor.getInstance();
+
+ @Builder.Default
+ private static final ParquetPartitionHelper parquetPartitionHelper =
+ ParquetPartitionHelper.getInstance();
+
+ private Map<String, List<String>> initPartitionInfo() {
+ return fsHelper.getPartitionFromDirectoryStructure(
+ hadoopConf, basePath, Collections.emptyMap());
+ }
+
+ /**
+ * To infer schema getting the latest file assumption is that latest file will have new fields
+ *
+ * @param modificationTime the commit to consider for reading the table state
+ * @return
+ */
+ @Override
+ public InternalTable getTable(Long modificationTime) {
+
+ Optional<LocatedFileStatus> latestFile =
+ fsHelper
+ .getParquetFiles(hadoopConf, basePath)
+ .max(Comparator.comparing(FileStatus::getModificationTime));
+
+ ParquetMetadata parquetMetadata =
+ parquetMetadataExtractor.readParquetMetadata(hadoopConf, latestFile.get().getPath());
+ Schema tableSchema =
+ new org.apache.parquet.avro.AvroSchemaConverter()
+ .convert(parquetMetadata.getFileMetaData().getSchema());
+
+ Set<String> partitionKeys = initPartitionInfo().keySet();
+
+ // merge schema of partition into original as partition is not part of parquet fie
+ if (!partitionKeys.isEmpty()) {
+ tableSchema = mergeAvroSchema(tableSchema, partitionKeys);
+ }
+ InternalSchema schema = schemaExtractor.toInternalSchema(tableSchema);
+
+ List<InternalPartitionField> partitionFields =
+ partitionKeys.isEmpty()
+ ? Collections.emptyList()
+ : parquetPartitionHelper.getInternalPartitionField(partitionKeys, schema);
+ DataLayoutStrategy dataLayoutStrategy =
+ partitionFields.isEmpty()
+ ? DataLayoutStrategy.FLAT
+ : DataLayoutStrategy.HIVE_STYLE_PARTITION;
+ return InternalTable.builder()
+ .tableFormat(TableFormat.PARQUET)
+ .basePath(basePath)
+ .name(tableName)
+ .layoutStrategy(dataLayoutStrategy)
+ .partitioningFields(partitionFields)
+ .readSchema(schema)
+ .latestCommitTime(Instant.ofEpochMilli(latestFile.get().getModificationTime()))
+ .build();
+ }
+
+ /**
+ * Here to get current snapshot listing all files hence the -1 is being passed
+ *
+ * @return
+ */
+ @Override
+ public InternalSnapshot getCurrentSnapshot() {
+
+ List<LocatedFileStatus> latestFile =
+ fsHelper.getParquetFiles(hadoopConf, basePath).collect(Collectors.toList());
+ Map<String, List<String>> partitionInfo = initPartitionInfo();
+ InternalTable table = getTable(-1L);
+ List<InternalDataFile> internalDataFiles =
+ latestFile.stream()
+ .map(
+ file ->
+ InternalDataFile.builder()
Review Comment:
Can the logic for this conversion move to a common method that can also be called from the `getTableChangeForCommit`?
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sync parquet to other file formats [incubator-xtable]
Posted by "sudhar91 (via GitHub)" <gi...@apache.org>.
sudhar91 commented on code in PR #592:
URL: https://github.com/apache/incubator-xtable/pull/592#discussion_r1885712084
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java:
##########
@@ -0,0 +1,209 @@
+package org.apache.xtable.parquet;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.*;
+import java.util.stream.Collectors;
+import lombok.Builder;
+import lombok.NonNull;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.xtable.avro.AvroSchemaConverter;
+import org.apache.xtable.model.*;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.*;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Builder
+public class ParquetConversionSource implements ConversionSource<Long> {
+
+ private final String tableName;
+ private final String basePath;
+ @NonNull private final Configuration hadoopConf;
+
+ @Builder.Default
+ private static final AvroSchemaConverter schemaExtractor = AvroSchemaConverter.getInstance();
+
+ @Builder.Default private static final FileSystemHelper fsHelper = FileSystemHelper.getInstance();
+
+ @Builder.Default
+ private static final ParquetMetadataExtractor parquetMetadataExtractor =
+ ParquetMetadataExtractor.getInstance();
+
+ @Builder.Default
+ private static final ParquetPartitionHelper parquetPartitionHelper =
+ ParquetPartitionHelper.getInstance();
+
+ private Map<String, List<String>> initPartitionInfo() {
+ return fsHelper.getPartitionFromDirectoryStructure(
+ hadoopConf, basePath, Collections.emptyMap());
+ }
+
+ /**
+ * To infer schema getting the latest file assumption is that latest file will have new fields
+ *
+ * @param modificationTime the commit to consider for reading the table state
+ * @return
+ */
+ @Override
+ public InternalTable getTable(Long modificationTime) {
+
+ Optional<LocatedFileStatus> latestFile =
+ fsHelper
+ .getParquetFiles(hadoopConf, basePath)
+ .max(Comparator.comparing(FileStatus::getModificationTime));
Review Comment:
i dont think we can push down with the api , even to filter files greater than modification we have to first list and then filter out, do u have any other idea on your mind for this?
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sync parquet to other file formats [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on PR #592:
URL: https://github.com/apache/incubator-xtable/pull/592#issuecomment-2526369203
@sudhar91 this is a great step forward on this feature! My main request is to pull some of these classes that are focused on conversion such as the partition and data file conversion into their own PR with unit tests written. It will be easier and quicker to get those straightforward changes reviewed while the rest of the details are figured out.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sync parquet to other file formats [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #592:
URL: https://github.com/apache/incubator-xtable/pull/592#discussion_r1895238705
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java:
##########
@@ -0,0 +1,209 @@
+package org.apache.xtable.parquet;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.*;
+import java.util.stream.Collectors;
+import lombok.Builder;
+import lombok.NonNull;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.xtable.avro.AvroSchemaConverter;
+import org.apache.xtable.model.*;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.*;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Builder
+public class ParquetConversionSource implements ConversionSource<Long> {
+
+ private final String tableName;
+ private final String basePath;
+ @NonNull private final Configuration hadoopConf;
+
+ @Builder.Default
+ private static final AvroSchemaConverter schemaExtractor = AvroSchemaConverter.getInstance();
+
+ @Builder.Default private static final FileSystemHelper fsHelper = FileSystemHelper.getInstance();
+
+ @Builder.Default
+ private static final ParquetMetadataExtractor parquetMetadataExtractor =
+ ParquetMetadataExtractor.getInstance();
+
+ @Builder.Default
+ private static final ParquetPartitionHelper parquetPartitionHelper =
+ ParquetPartitionHelper.getInstance();
+
+ private Map<String, List<String>> initPartitionInfo() {
+ return fsHelper.getPartitionFromDirectoryStructure(
+ hadoopConf, basePath, Collections.emptyMap());
+ }
+
+ /**
+ * To infer schema getting the latest file assumption is that latest file will have new fields
+ *
+ * @param modificationTime the commit to consider for reading the table state
+ * @return
+ */
+ @Override
+ public InternalTable getTable(Long modificationTime) {
+
+ Optional<LocatedFileStatus> latestFile =
+ fsHelper
+ .getParquetFiles(hadoopConf, basePath)
+ .max(Comparator.comparing(FileStatus::getModificationTime));
Review Comment:
No, just wanted to see if it was possible to help with large tables
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sync parquet to other file formats [incubator-xtable]
Posted by "sudhar91 (via GitHub)" <gi...@apache.org>.
sudhar91 commented on code in PR #592:
URL: https://github.com/apache/incubator-xtable/pull/592#discussion_r1885715183
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java:
##########
@@ -0,0 +1,209 @@
+package org.apache.xtable.parquet;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.*;
+import java.util.stream.Collectors;
+import lombok.Builder;
+import lombok.NonNull;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.xtable.avro.AvroSchemaConverter;
+import org.apache.xtable.model.*;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.*;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Builder
+public class ParquetConversionSource implements ConversionSource<Long> {
Review Comment:
Can u explain a bit on the interval and how are u envisioning ? this long is the last synced modification time of file so in next run list files greater mod time so new files are synced.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sync parquet to other file formats [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #592:
URL: https://github.com/apache/incubator-xtable/pull/592#discussion_r1895243576
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java:
##########
@@ -0,0 +1,209 @@
+package org.apache.xtable.parquet;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.*;
+import java.util.stream.Collectors;
+import lombok.Builder;
+import lombok.NonNull;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.xtable.avro.AvroSchemaConverter;
+import org.apache.xtable.model.*;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.*;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Builder
+public class ParquetConversionSource implements ConversionSource<Long> {
Review Comment:
I was thinking of an interval since it can also easily show where the start time was for the sync. This could be useful when the targets fall out of sync with each other. Currently if there are commits `1`, `2`, and `3` in the source and `Target1` only is synced to `1` but `Target2` is synced to `2`, the incremental sync can sync `2` and `3` to `Target1` and only `3` to `Target2` as part of the same sync. I am not sure what that will look like for this source so I was thinking intervals can help us define these "commits" but I need to think through it some more.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sync parquet to other file formats [incubator-xtable]
Posted by "sudhar91 (via GitHub)" <gi...@apache.org>.
sudhar91 commented on code in PR #592:
URL: https://github.com/apache/incubator-xtable/pull/592#discussion_r1885714008
##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java:
##########
@@ -0,0 +1,209 @@
+package org.apache.xtable.parquet;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.*;
+import java.util.stream.Collectors;
+import lombok.Builder;
+import lombok.NonNull;
+import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.xtable.avro.AvroSchemaConverter;
+import org.apache.xtable.model.*;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.*;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Builder
+public class ParquetConversionSource implements ConversionSource<Long> {
+
+ private final String tableName;
+ private final String basePath;
+ @NonNull private final Configuration hadoopConf;
+
+ @Builder.Default
+ private static final AvroSchemaConverter schemaExtractor = AvroSchemaConverter.getInstance();
+
+ @Builder.Default private static final FileSystemHelper fsHelper = FileSystemHelper.getInstance();
+
+ @Builder.Default
+ private static final ParquetMetadataExtractor parquetMetadataExtractor =
Review Comment:
i added initially but not using it will remove it
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Sync parquet to other file formats [incubator-xtable]
Posted by "sudhar91 (via GitHub)" <gi...@apache.org>.
sudhar91 commented on PR #592:
URL: https://github.com/apache/incubator-xtable/pull/592#issuecomment-2543886513
> @sudhar91 this is a great step forward on this feature! My main request is to pull some of these classes that are focused on conversion such as the partition and data file conversion into their own PR with unit tests written. It will be easier and quicker to get those straightforward changes reviewed while the rest of the details are figured out.
Thanks for taking ur time to review , i couldnt push my unit test because i was having a weird jar conflicts on my test will fix them next week, i raised this PR because wanted to validate whether my approach is right in terms of this feature :)
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org