You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@xtable.apache.org by "mikedias (via GitHub)" <gi...@apache.org> on 2025/09/08 05:46:03 UTC
[PR] [Draft] Paimon Source Support [incubator-xtable]
mikedias opened a new pull request, #742:
URL: https://github.com/apache/incubator-xtable/pull/742
Issue: https://github.com/apache/incubator-xtable/issues/275
Draft PR to add support for [Apache Paimon](https://paimon.apache.org/) as the source table. This is an early sharing to collect feedback on whether the approach is right and on a few issues I've encountered. 🙂
Things that are missing yet:
- **Comprehensive unit test:** I've relied on `ITConversionController` to validate the code so far. Once we are good with the approach, I'll cover the implementation with more tests.
- **Incremental Sync:** I've only implemented the `ConversionSource` methods for the snapshot sync. Once we are good with the approach, I'll implement the incremental sync methods.
- **Target to Paimon:** I've only focused on implementing Paimon as a source. The target implementation will be out of scope for this contribution, if that's okay.
Things where I need help:
### **Hudi partitions and Paimon buckets**
Paimon has buckets as another folder division within partitions (e.g. `partition=2025-09-01/bucket-0/data-file.parquet`), however, Hudi considers the bucket value as part of the partition value (e.g. `2025-09-01/bucket-0`). Looking at the code, it seems that directory structure == partition values assumption is pretty deep, so I wonder if there is a way to work around that, or we should call out as a limitation between both formats.
### **Iceberg Parquet conversion errors**
When executing the `ITConversionController#testVariousOperations` with source=paimon and target=iceberg, I'm facing the follow error:
```
class org.apache.iceberg.shaded.org.apache.arrow.vector.IntVector cannot be cast to class org.apache.iceberg.shaded.org.apache.arrow.vector.BaseVariableWidthVector
```
From debugging, it seems the Parquet reader is trying to read the `id int` field value with the `VarWidthReader` class instead of the `IntegerReader`, causing the conversion issue:
<img width="512" height="173" alt="image" src="https://github.com/user-attachments/assets/d4f98940-e0f9-4ba2-9d57-2a1403457a02" />
Disabling vectorization doesn't help; doing so makes the error appear on the Spark level, indicating that there is something wrong with the schema conversion for Iceberg that I quite don't understand... 🤔
Thank you so much in advance for your help!
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2412293433
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
Amazing, thank you so much @the-other-tim-brown!!! 😄
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2330344552
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonConversionSourceProvider.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.paimon;
+
+import java.io.IOException;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+public class PaimonConversionSourceProvider extends ConversionSourceProvider<Snapshot> {
+ @Override
+ public ConversionSource<Snapshot> getConversionSourceInstance(SourceTable sourceTableConfig) {
+ try {
+ Options catalogOptions = new Options();
+ CatalogContext context = CatalogContext.create(catalogOptions, hadoopConf);
+
+ Path path = new Path(sourceTableConfig.getDataPath());
+ FileIO fileIO = FileIO.get(path, context);
+ FileStoreTable paimonTable = FileStoreTableFactory.create(fileIO, path);
+
+ return new PaimonConversionSource(paimonTable);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
Review Comment:
We have some custom exception types like `ReadException` that may be better here to indicate there is an issue reading the initial state.
##########
xtable-core/pom.xml:
##########
@@ -206,6 +218,14 @@
<skip>false</skip>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>11</source>
+ <target>11</target>
Review Comment:
Is this a requirement for Paimon? We're only publishing to a java 8 target currently
##########
xtable-core/src/test/java/org/apache/xtable/ITConversionController.java:
##########
@@ -37,16 +35,7 @@
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Base64;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
-import java.util.UUID;
+import java.util.*;
Review Comment:
nitpick: generally we try to avoid `*` imports.
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
Adding a note here likely related to the issues with Iceberg you are seeing in the ITs. The output of this does not include the meta fields `_KEY_id`, `_SEQUENCE_NUMBER`, `_VALUE_KIND`. This is likely messing up the type to offset mapping in Iceberg. Is there a way to extract the Paimon schema with these fields?
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
+ private static final PaimonSchemaExtractor INSTANCE = new PaimonSchemaExtractor();
+
+ public static PaimonSchemaExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public InternalSchema toInternalSchema(TableSchema paimonSchema) {
+ RowType rowType = paimonSchema.logicalRowType();
+ List<InternalField> fields = toInternalFields(rowType);
+ return InternalSchema.builder()
+ .name("record")
+ .dataType(InternalType.RECORD)
+ .fields(fields)
+ .recordKeyFields(primaryKeyFields(paimonSchema, fields))
+ .build();
+ }
+
+ private List<InternalField> primaryKeyFields(
+ TableSchema paimonSchema, List<InternalField> internalFields) {
+ List<String> keys = paimonSchema.primaryKeys();
Review Comment:
Is it possible for the primary key to be a nested field in Paimon?
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2333615594
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
Does Paimon attach the field index to the parquet data? The field ordering should match that. We have an option to pass in the index directly to the internal fields as well.
If not, you will not be able to read the data in some vendors like BigQuery and Snowflake.
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2331973890
##########
xtable-core/pom.xml:
##########
@@ -206,6 +218,14 @@
<skip>false</skip>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>11</source>
+ <target>11</target>
Review Comment:
I'm getting these compatibility errors when using java 8 on my machine, so I thought xtable was on 11 already but wasn't configured
<img width="1105" height="161" alt="image" src="https://github.com/user-attachments/assets/0c9d224a-a729-4381-ae40-4e89eb8e265b" />
<img width="1639" height="96" alt="image" src="https://github.com/user-attachments/assets/3d25d9f3-b1fe-4548-806b-756af482bf71" />
everything works fine on 11 🤔
--
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] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2484525039
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonPartitionExtractor.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.paimon;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
+
+import org.apache.xtable.model.schema.InternalField;
+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;
+
+/** Extracts partition spec for Paimon as identity transforms on partition keys. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonPartitionExtractor {
+
+ private final PaimonSchemaExtractor paimonSchemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private static final PaimonPartitionExtractor INSTANCE = new PaimonPartitionExtractor();
+
+ public static PaimonPartitionExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalPartitionField> toInternalPartitionFields(
+ List<String> partitionKeys, InternalSchema schema) {
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return partitionKeys.stream()
+ .map(key -> toPartitionField(key, schema))
+ .collect(Collectors.toList());
+ }
+
+ public List<PartitionValue> toPartitionValues(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ InternalSchema internalSchema = paimonSchemaExtractor.toInternalSchema(table.schema());
+
+ List<PartitionValue> partitionValues = new ArrayList<>();
+ for (Map.Entry<String, String> entry :
+ partitionComputer.generatePartValues(partition).entrySet()) {
+ PartitionValue partitionValue =
+ PartitionValue.builder()
+ .partitionField(toPartitionField(entry.getKey(), internalSchema))
+ .range(Range.scalar(entry.getValue()))
+ .build();
+ partitionValues.add(partitionValue);
+ }
+ return partitionValues;
+ }
+
+ public Optional<String> toPartitionPath(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ return partitionComputer.generatePartValues(partition).entrySet().stream()
+ .map(e -> e.getKey() + "=" + e.getValue())
+ .reduce((a, b) -> a + "/" + b);
+ }
+
+ private InternalPartitionField toPartitionField(String key, InternalSchema schema) {
+ InternalField sourceField =
+ findField(schema, key)
+ .orElseThrow(
+ () -> new IllegalArgumentException("Partition key not found in schema: " + key));
+ return InternalPartitionField.builder()
+ .sourceField(sourceField)
+ .transformType(PartitionTransformType.VALUE)
Review Comment:
yes, and it also allows some customization https://paimon.apache.org/docs/1.2/maintenance/configurations/
<img width="863" height="506" alt="image" src="https://github.com/user-attachments/assets/51011cc0-97c5-417c-aeb6-88f20a59de86" />
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2409089306
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
They are [defined here](https://github.com/apache/paimon/blob/daf482e2866cd0ec7c24700b27261d19a60106f9/paimon-api/src/main/java/org/apache/paimon/table/SpecialFields.java#L79) and [set here](https://github.com/apache/paimon/blob/daf482e2866cd0ec7c24700b27261d19a60106f9/paimon-core/src/main/java/org/apache/paimon/KeyValue.java#L134). Is that what you are asking?
--
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] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#issuecomment-3476468230
> @the-other-tim-brown I've added the missing unit tests, and I think it's good to review. Let me know if there is something I'm missing or you want to approach differently.
@mikedias apologies for the delay on this. Reviewing it now.
--
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] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2483771095
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonConversionSource.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.paimon;
+
+import static org.apache.xtable.model.storage.DataLayoutStrategy.HIVE_STYLE_PARTITION;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.List;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.xtable.exception.ReadException;
+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.DataLayoutStrategy;
+import org.apache.xtable.model.storage.InternalDataFile;
+import org.apache.xtable.model.storage.PartitionFileGroup;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Log4j2
+public class PaimonConversionSource implements ConversionSource<Snapshot> {
+
+ private final FileStoreTable paimonTable;
+ private final SchemaManager schemaManager;
+ private final SnapshotManager snapshotManager;
+
+ private final PaimonDataFileExtractor dataFileExtractor = PaimonDataFileExtractor.getInstance();
+ private final PaimonSchemaExtractor schemaExtractor = PaimonSchemaExtractor.getInstance();
+ private final PaimonPartitionExtractor partitionSpecExtractor =
+ PaimonPartitionExtractor.getInstance();
+
+ public PaimonConversionSource(FileStoreTable paimonTable) {
+ this.paimonTable = paimonTable;
+ this.schemaManager = paimonTable.schemaManager();
+ this.snapshotManager = paimonTable.snapshotManager();
+ }
+
+ @Override
+ public InternalTable getTable(Snapshot snapshot) {
+ TableSchema paimonSchema = schemaManager.schema(snapshot.schemaId());
+ InternalSchema internalSchema = schemaExtractor.toInternalSchema(paimonSchema);
+
+ List<String> partitionKeys = paimonTable.partitionKeys();
+ List<InternalPartitionField> partitioningFields =
+ partitionSpecExtractor.toInternalPartitionFields(partitionKeys, internalSchema);
+
+ DataLayoutStrategy dataLayoutStrategy =
+ partitioningFields.isEmpty() ? DataLayoutStrategy.FLAT : HIVE_STYLE_PARTITION;
+
+ return InternalTable.builder()
+ .name(paimonTable.name())
+ .tableFormat(TableFormat.PAIMON)
+ .readSchema(internalSchema)
+ .layoutStrategy(dataLayoutStrategy)
+ .basePath(paimonTable.location().toString())
+ .partitioningFields(partitioningFields)
+ .latestCommitTime(Instant.ofEpochMilli(snapshot.timeMillis()))
+ .latestMetadataPath(snapshotManager.snapshotPath(snapshot.id()).toString())
+ .build();
+ }
+
+ @Override
+ public InternalTable getCurrentTable() {
+ SnapshotManager snapshotManager = paimonTable.snapshotManager();
+ Snapshot snapshot = snapshotManager.latestSnapshot();
+ if (snapshot == null) {
+ throw new ReadException("No snapshots found for table " + paimonTable.name());
+ }
Review Comment:
Nitpick: can we move this code and lines 99-103 into a little private, helper method
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonConversionSource.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.paimon;
+
+import static org.apache.xtable.model.storage.DataLayoutStrategy.HIVE_STYLE_PARTITION;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.List;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.xtable.exception.ReadException;
+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.DataLayoutStrategy;
+import org.apache.xtable.model.storage.InternalDataFile;
+import org.apache.xtable.model.storage.PartitionFileGroup;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Log4j2
+public class PaimonConversionSource implements ConversionSource<Snapshot> {
+
+ private final FileStoreTable paimonTable;
+ private final SchemaManager schemaManager;
+ private final SnapshotManager snapshotManager;
+
+ private final PaimonDataFileExtractor dataFileExtractor = PaimonDataFileExtractor.getInstance();
+ private final PaimonSchemaExtractor schemaExtractor = PaimonSchemaExtractor.getInstance();
+ private final PaimonPartitionExtractor partitionSpecExtractor =
+ PaimonPartitionExtractor.getInstance();
+
+ public PaimonConversionSource(FileStoreTable paimonTable) {
+ this.paimonTable = paimonTable;
+ this.schemaManager = paimonTable.schemaManager();
+ this.snapshotManager = paimonTable.snapshotManager();
+ }
+
+ @Override
+ public InternalTable getTable(Snapshot snapshot) {
+ TableSchema paimonSchema = schemaManager.schema(snapshot.schemaId());
+ InternalSchema internalSchema = schemaExtractor.toInternalSchema(paimonSchema);
+
+ List<String> partitionKeys = paimonTable.partitionKeys();
+ List<InternalPartitionField> partitioningFields =
+ partitionSpecExtractor.toInternalPartitionFields(partitionKeys, internalSchema);
+
+ DataLayoutStrategy dataLayoutStrategy =
+ partitioningFields.isEmpty() ? DataLayoutStrategy.FLAT : HIVE_STYLE_PARTITION;
Review Comment:
Does Paimon always include the field name in the path or is there an option to not do this?
For example, Hive style means date would be `base/path/data=2025-11-01/data` and not simply `base/path/2025-11-01/data`
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonDataFileExtractor.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.paimon;
+
+import java.util.*;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+
+import org.apache.xtable.model.stat.ColumnStat;
+import org.apache.xtable.model.storage.InternalDataFile;
+
+public class PaimonDataFileExtractor {
+
+ private final PaimonPartitionExtractor partitionExtractor =
+ PaimonPartitionExtractor.getInstance();
+
+ private static final PaimonDataFileExtractor INSTANCE = new PaimonDataFileExtractor();
+
+ public static PaimonDataFileExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalDataFile> toInternalDataFiles(FileStoreTable table, Snapshot snapshot) {
+ List<InternalDataFile> result = new ArrayList<>();
+ Iterator<ManifestEntry> manifestEntryIterator =
+ newSnapshotReader(table, snapshot).readFileIterator();
+ while (manifestEntryIterator.hasNext()) {
+ result.add(toInternalDataFile(table, manifestEntryIterator.next()));
+ }
+ return result;
+ }
+
+ private InternalDataFile toInternalDataFile(FileStoreTable table, ManifestEntry entry) {
+ return InternalDataFile.builder()
+ .physicalPath(toFullPhysicalPath(table, entry))
+ .fileSizeBytes(entry.file().fileSize())
+ .lastModified(entry.file().creationTimeEpochMillis())
+ .recordCount(entry.file().rowCount())
+ .partitionValues(partitionExtractor.toPartitionValues(table, entry.partition()))
+ .columnStats(toColumnStats(entry.file()))
+ .build();
+ }
+
+ private String toFullPhysicalPath(FileStoreTable table, ManifestEntry entry) {
+ String basePath = table.location().toString();
+ String bucketPath = "bucket-" + entry.bucket();
+ String filePath = entry.file().fileName();
+
+ Optional<String> partitionPath = partitionExtractor.toPartitionPath(table, entry.partition());
+ if (partitionPath.isPresent()) {
+ return String.join("/", basePath, partitionPath.get(), bucketPath, filePath);
+ } else {
+ return String.join("/", basePath, bucketPath, filePath);
+ }
+ }
+
+ private List<ColumnStat> toColumnStats(DataFileMeta file) {
+ // TODO: Implement logic to extract column stats from the file meta
Review Comment:
Let's track this as a separate GH issue if we are not already.
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonConversionSourceProvider.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.paimon;
+
+import java.io.IOException;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.exception.ReadException;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+public class PaimonConversionSourceProvider extends ConversionSourceProvider<Snapshot> {
+ @Override
+ public ConversionSource<Snapshot> getConversionSourceInstance(SourceTable sourceTableConfig) {
+ try {
+ Options catalogOptions = new Options();
+ CatalogContext context = CatalogContext.create(catalogOptions, hadoopConf);
+
+ Path path = new Path(sourceTableConfig.getDataPath());
+ FileIO fileIO = FileIO.get(path, context);
+ FileStoreTable paimonTable = FileStoreTableFactory.create(fileIO, path);
+
+ return new PaimonConversionSource(paimonTable);
+ } catch (IOException e) {
+ throw new ReadException(e.getMessage());
Review Comment:
Let's update this handling so the stacktrace is propagated.
`throw new ReadException("Failed to read Paimon table from file system", e)`
##########
xtable-core/src/test/java/org/apache/xtable/ITConversionController.java:
##########
@@ -104,9 +106,12 @@
import org.apache.xtable.iceberg.TestIcebergDataHelper;
import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.model.sync.SyncMode;
+import org.apache.xtable.paimon.PaimonConversionSourceProvider;
public class ITConversionController {
- @TempDir public static Path tempDir;
+ @TempDir(cleanup = CleanupMode.NEVER) // TODO remove CleanupMode.NEVER after debugging
Review Comment:
Can this be reverted now?
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonDataFileExtractor.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.paimon;
+
+import java.util.*;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+
+import org.apache.xtable.model.stat.ColumnStat;
+import org.apache.xtable.model.storage.InternalDataFile;
+
+public class PaimonDataFileExtractor {
+
+ private final PaimonPartitionExtractor partitionExtractor =
+ PaimonPartitionExtractor.getInstance();
+
+ private static final PaimonDataFileExtractor INSTANCE = new PaimonDataFileExtractor();
+
+ public static PaimonDataFileExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalDataFile> toInternalDataFiles(FileStoreTable table, Snapshot snapshot) {
+ List<InternalDataFile> result = new ArrayList<>();
+ Iterator<ManifestEntry> manifestEntryIterator =
+ newSnapshotReader(table, snapshot).readFileIterator();
+ while (manifestEntryIterator.hasNext()) {
+ result.add(toInternalDataFile(table, manifestEntryIterator.next()));
+ }
+ return result;
+ }
+
+ private InternalDataFile toInternalDataFile(FileStoreTable table, ManifestEntry entry) {
+ return InternalDataFile.builder()
+ .physicalPath(toFullPhysicalPath(table, entry))
+ .fileSizeBytes(entry.file().fileSize())
+ .lastModified(entry.file().creationTimeEpochMillis())
+ .recordCount(entry.file().rowCount())
+ .partitionValues(partitionExtractor.toPartitionValues(table, entry.partition()))
+ .columnStats(toColumnStats(entry.file()))
+ .build();
+ }
+
+ private String toFullPhysicalPath(FileStoreTable table, ManifestEntry entry) {
+ String basePath = table.location().toString();
+ String bucketPath = "bucket-" + entry.bucket();
+ String filePath = entry.file().fileName();
+
+ Optional<String> partitionPath = partitionExtractor.toPartitionPath(table, entry.partition());
+ if (partitionPath.isPresent()) {
+ return String.join("/", basePath, partitionPath.get(), bucketPath, filePath);
+ } else {
+ return String.join("/", basePath, bucketPath, filePath);
+ }
+ }
+
+ private List<ColumnStat> toColumnStats(DataFileMeta file) {
+ // TODO: Implement logic to extract column stats from the file meta
+ return Collections.emptyList();
+ }
+
+ private SnapshotReader newSnapshotReader(FileStoreTable table, Snapshot snapshot) {
+ // If the table has primary keys, we read only the top level files
Review Comment:
Just curious, is this similar to the Hudi Merge on Read table?
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonConversionSourceProvider.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.paimon;
+
+import java.io.IOException;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.exception.ReadException;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+public class PaimonConversionSourceProvider extends ConversionSourceProvider<Snapshot> {
+ @Override
+ public ConversionSource<Snapshot> getConversionSourceInstance(SourceTable sourceTableConfig) {
+ try {
+ Options catalogOptions = new Options();
Review Comment:
We don't need any changes right now but will the user want to supply some custom options here?
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonConversionSource.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.paimon;
+
+import static org.apache.xtable.model.storage.DataLayoutStrategy.HIVE_STYLE_PARTITION;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.List;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.xtable.exception.ReadException;
+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.DataLayoutStrategy;
+import org.apache.xtable.model.storage.InternalDataFile;
+import org.apache.xtable.model.storage.PartitionFileGroup;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Log4j2
+public class PaimonConversionSource implements ConversionSource<Snapshot> {
+
+ private final FileStoreTable paimonTable;
+ private final SchemaManager schemaManager;
+ private final SnapshotManager snapshotManager;
+
+ private final PaimonDataFileExtractor dataFileExtractor = PaimonDataFileExtractor.getInstance();
+ private final PaimonSchemaExtractor schemaExtractor = PaimonSchemaExtractor.getInstance();
+ private final PaimonPartitionExtractor partitionSpecExtractor =
+ PaimonPartitionExtractor.getInstance();
+
+ public PaimonConversionSource(FileStoreTable paimonTable) {
+ this.paimonTable = paimonTable;
+ this.schemaManager = paimonTable.schemaManager();
+ this.snapshotManager = paimonTable.snapshotManager();
+ }
+
+ @Override
+ public InternalTable getTable(Snapshot snapshot) {
+ TableSchema paimonSchema = schemaManager.schema(snapshot.schemaId());
+ InternalSchema internalSchema = schemaExtractor.toInternalSchema(paimonSchema);
+
+ List<String> partitionKeys = paimonTable.partitionKeys();
+ List<InternalPartitionField> partitioningFields =
+ partitionSpecExtractor.toInternalPartitionFields(partitionKeys, internalSchema);
+
+ DataLayoutStrategy dataLayoutStrategy =
+ partitioningFields.isEmpty() ? DataLayoutStrategy.FLAT : HIVE_STYLE_PARTITION;
+
+ return InternalTable.builder()
+ .name(paimonTable.name())
+ .tableFormat(TableFormat.PAIMON)
+ .readSchema(internalSchema)
+ .layoutStrategy(dataLayoutStrategy)
+ .basePath(paimonTable.location().toString())
+ .partitioningFields(partitioningFields)
+ .latestCommitTime(Instant.ofEpochMilli(snapshot.timeMillis()))
+ .latestMetadataPath(snapshotManager.snapshotPath(snapshot.id()).toString())
+ .build();
+ }
+
+ @Override
+ public InternalTable getCurrentTable() {
+ SnapshotManager snapshotManager = paimonTable.snapshotManager();
+ Snapshot snapshot = snapshotManager.latestSnapshot();
+ if (snapshot == null) {
+ throw new ReadException("No snapshots found for table " + paimonTable.name());
+ }
+ return getTable(snapshot);
+ }
+
+ @Override
+ public InternalSnapshot getCurrentSnapshot() {
+ SnapshotManager snapshotManager = paimonTable.snapshotManager();
+ Snapshot snapshot = snapshotManager.latestSnapshot();
+ if (snapshot == null) {
+ throw new ReadException("No snapshots found for table " + paimonTable.name());
+ }
+
+ InternalTable internalTable = getTable(snapshot);
+ List<InternalDataFile> internalDataFiles =
+ dataFileExtractor.toInternalDataFiles(paimonTable, snapshot);
+
+ return InternalSnapshot.builder()
+ .table(internalTable)
+ .version(Long.toString(snapshot.timeMillis()))
+ .partitionedDataFiles(PartitionFileGroup.fromFiles(internalDataFiles))
+ // TODO : Implement pending commits extraction, required for incremental sync
Review Comment:
Nitpick: Can you create a GH Issue to track the incremental sync work if one does not already exist?
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonPartitionExtractor.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.paimon;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
+
+import org.apache.xtable.model.schema.InternalField;
+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;
+
+/** Extracts partition spec for Paimon as identity transforms on partition keys. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonPartitionExtractor {
+
+ private final PaimonSchemaExtractor paimonSchemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private static final PaimonPartitionExtractor INSTANCE = new PaimonPartitionExtractor();
+
+ public static PaimonPartitionExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalPartitionField> toInternalPartitionFields(
+ List<String> partitionKeys, InternalSchema schema) {
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return partitionKeys.stream()
+ .map(key -> toPartitionField(key, schema))
+ .collect(Collectors.toList());
+ }
+
+ public List<PartitionValue> toPartitionValues(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ InternalSchema internalSchema = paimonSchemaExtractor.toInternalSchema(table.schema());
Review Comment:
Can we pass in the schema to avoid this conversion per file?
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonPartitionExtractor.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.paimon;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
+
+import org.apache.xtable.model.schema.InternalField;
+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;
+
+/** Extracts partition spec for Paimon as identity transforms on partition keys. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonPartitionExtractor {
+
+ private final PaimonSchemaExtractor paimonSchemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private static final PaimonPartitionExtractor INSTANCE = new PaimonPartitionExtractor();
+
+ public static PaimonPartitionExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalPartitionField> toInternalPartitionFields(
+ List<String> partitionKeys, InternalSchema schema) {
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return partitionKeys.stream()
+ .map(key -> toPartitionField(key, schema))
+ .collect(Collectors.toList());
+ }
+
+ public List<PartitionValue> toPartitionValues(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ InternalSchema internalSchema = paimonSchemaExtractor.toInternalSchema(table.schema());
+
+ List<PartitionValue> partitionValues = new ArrayList<>();
Review Comment:
Nitpick: let's establish the list with the size of the map produced by `partitionComputer.generatePartValues(partition)` - this is done per file so we want to make sure we're as lean as possible when running the initial snapshot syncs on very large tables.
We can also return Collections.emptyList when the table is not partitioned.
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonPartitionExtractor.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.paimon;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
+
+import org.apache.xtable.model.schema.InternalField;
+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;
+
+/** Extracts partition spec for Paimon as identity transforms on partition keys. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonPartitionExtractor {
+
+ private final PaimonSchemaExtractor paimonSchemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private static final PaimonPartitionExtractor INSTANCE = new PaimonPartitionExtractor();
+
+ public static PaimonPartitionExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalPartitionField> toInternalPartitionFields(
+ List<String> partitionKeys, InternalSchema schema) {
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return partitionKeys.stream()
+ .map(key -> toPartitionField(key, schema))
+ .collect(Collectors.toList());
+ }
+
+ public List<PartitionValue> toPartitionValues(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ InternalSchema internalSchema = paimonSchemaExtractor.toInternalSchema(table.schema());
+
+ List<PartitionValue> partitionValues = new ArrayList<>();
+ for (Map.Entry<String, String> entry :
+ partitionComputer.generatePartValues(partition).entrySet()) {
+ PartitionValue partitionValue =
+ PartitionValue.builder()
+ .partitionField(toPartitionField(entry.getKey(), internalSchema))
+ .range(Range.scalar(entry.getValue()))
+ .build();
+ partitionValues.add(partitionValue);
+ }
+ return partitionValues;
+ }
+
+ public Optional<String> toPartitionPath(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ return partitionComputer.generatePartValues(partition).entrySet().stream()
+ .map(e -> e.getKey() + "=" + e.getValue())
+ .reduce((a, b) -> a + "/" + b);
+ }
+
+ private InternalPartitionField toPartitionField(String key, InternalSchema schema) {
+ InternalField sourceField =
+ findField(schema, key)
+ .orElseThrow(
+ () -> new IllegalArgumentException("Partition key not found in schema: " + key));
+ return InternalPartitionField.builder()
+ .sourceField(sourceField)
+ .transformType(PartitionTransformType.VALUE)
Review Comment:
Does Paimon allow users to partition on a timestamp?
##########
xtable-core/src/test/java/org/apache/xtable/paimon/TestPaimonSchemaExtractor.java:
##########
@@ -0,0 +1,496 @@
+/*
+ * 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.paimon;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.CharType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DateType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.FloatType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.SmallIntType;
+import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.types.TinyIntType;
+import org.apache.paimon.types.VarCharType;
+import org.junit.jupiter.api.Test;
+
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+
+public class TestPaimonSchemaExtractor {
+ private static final PaimonSchemaExtractor schemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private void assertField(DataField paimonField, InternalField expectedInternalField) {
+ assertField(paimonField, expectedInternalField, Collections.emptyList());
+ }
+
+ private void assertField(
+ DataField paimonField, InternalField expectedInternalField, List<String> primaryKeys) {
+ TableSchema paimonSchema =
+ new TableSchema(
+ 0,
+ Collections.singletonList(paimonField),
+ 0,
+ Collections.emptyList(),
+ primaryKeys,
+ new HashMap<>(),
+ "");
+ InternalSchema internalSchema = schemaExtractor.toInternalSchema(paimonSchema);
+ List<InternalField> recordKeyFields =
+ primaryKeys.isEmpty()
+ ? Collections.emptyList()
+ : Collections.singletonList(expectedInternalField);
+ InternalSchema expectedSchema =
+ InternalSchema.builder()
+ .name("record")
+ .dataType(InternalType.RECORD)
+ .fields(Collections.singletonList(expectedInternalField))
+ .recordKeyFields(recordKeyFields)
+ .build();
+ assertEquals(expectedSchema, internalSchema);
+ }
+
+ @Test
+ void testCharField() {
+ DataField paimonField = new DataField(0, "char_field", new CharType(10));
+ InternalField expectedField =
+ InternalField.builder()
+ .name("char_field")
+ .fieldId(0)
+ .schema(
+ InternalSchema.builder()
+ .name("CHAR(10)")
+ .dataType(InternalType.STRING)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testVarcharField() {
+ DataField paimonField = new DataField(1, "varchar_field", new VarCharType(255));
+ InternalField expectedField =
+ InternalField.builder()
+ .name("varchar_field")
+ .fieldId(1)
+ .schema(
+ InternalSchema.builder()
+ .name("VARCHAR(255)")
+ .dataType(InternalType.STRING)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testBooleanField() {
+ DataField paimonField = new DataField(2, "boolean_field", new BooleanType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("boolean_field")
+ .fieldId(2)
+ .schema(
+ InternalSchema.builder()
+ .name("BOOLEAN")
+ .dataType(InternalType.BOOLEAN)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testTinyIntField() {
+ DataField paimonField = new DataField(3, "tinyint_field", new TinyIntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("tinyint_field")
+ .fieldId(3)
+ .schema(
+ InternalSchema.builder()
+ .name("TINYINT")
+ .dataType(InternalType.INT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testSmallIntField() {
+ DataField paimonField = new DataField(4, "smallint_field", new SmallIntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("smallint_field")
+ .fieldId(4)
+ .schema(
+ InternalSchema.builder()
+ .name("SMALLINT")
+ .dataType(InternalType.INT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testIntField() {
+ DataField paimonField = new DataField(5, "int_field", new IntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("int_field")
+ .fieldId(5)
+ .schema(
+ InternalSchema.builder()
+ .name("INT")
+ .dataType(InternalType.INT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testBigIntField() {
+ DataField paimonField = new DataField(6, "bigint_field", new BigIntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("bigint_field")
+ .fieldId(6)
+ .schema(
+ InternalSchema.builder()
+ .name("BIGINT")
+ .dataType(InternalType.LONG)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testFloatField() {
+ DataField paimonField = new DataField(7, "float_field", new FloatType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("float_field")
+ .fieldId(7)
+ .schema(
+ InternalSchema.builder()
+ .name("FLOAT")
+ .dataType(InternalType.FLOAT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testDoubleField() {
+ DataField paimonField = new DataField(8, "double_field", new DoubleType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("double_field")
+ .fieldId(8)
+ .schema(
+ InternalSchema.builder()
+ .name("DOUBLE")
+ .dataType(InternalType.DOUBLE)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testDateField() {
+ DataField paimonField = new DataField(9, "date_field", new DateType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("date_field")
+ .fieldId(9)
+ .schema(
+ InternalSchema.builder()
+ .name("DATE")
+ .dataType(InternalType.DATE)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testTimestampField() {
+ DataField paimonField = new DataField(10, "timestamp_field", new TimestampType(3));
+ Map<InternalSchema.MetadataKey, Object> timestampMetadata =
+ Collections.singletonMap(
+ InternalSchema.MetadataKey.TIMESTAMP_PRECISION, InternalSchema.MetadataValue.MICROS);
+ InternalField expectedField =
+ InternalField.builder()
+ .name("timestamp_field")
+ .fieldId(10)
+ .schema(
+ InternalSchema.builder()
+ .name("TIMESTAMP(3)")
+ .dataType(InternalType.TIMESTAMP)
+ .isNullable(true)
+ .metadata(timestampMetadata)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testDecimalField() {
+ DataField paimonField = new DataField(11, "decimal_field", new DecimalType(10, 2));
+ Map<InternalSchema.MetadataKey, Object> decimalMetadata = new HashMap<>();
+ decimalMetadata.put(InternalSchema.MetadataKey.DECIMAL_PRECISION, 10);
+ decimalMetadata.put(InternalSchema.MetadataKey.DECIMAL_SCALE, 2);
+ InternalField expectedField =
+ InternalField.builder()
+ .name("decimal_field")
+ .fieldId(11)
+ .schema(
+ InternalSchema.builder()
+ .name("DECIMAL(10, 2)")
+ .dataType(InternalType.DECIMAL)
+ .isNullable(true)
+ .metadata(decimalMetadata)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testStructField() {
Review Comment:
Nitpick: for my own sanity, I usually like to go three levels deep to ensure that the parent path is not simply the name of the parent field but rather the full path to that field. Can we do that here?
##########
xtable-core/src/test/java/org/apache/xtable/paimon/TestPaimonSchemaExtractor.java:
##########
@@ -0,0 +1,496 @@
+/*
+ * 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.paimon;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.CharType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DateType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.FloatType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.SmallIntType;
+import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.types.TinyIntType;
+import org.apache.paimon.types.VarCharType;
+import org.junit.jupiter.api.Test;
+
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+
+public class TestPaimonSchemaExtractor {
+ private static final PaimonSchemaExtractor schemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private void assertField(DataField paimonField, InternalField expectedInternalField) {
+ assertField(paimonField, expectedInternalField, Collections.emptyList());
+ }
+
+ private void assertField(
+ DataField paimonField, InternalField expectedInternalField, List<String> primaryKeys) {
+ TableSchema paimonSchema =
+ new TableSchema(
+ 0,
+ Collections.singletonList(paimonField),
+ 0,
+ Collections.emptyList(),
+ primaryKeys,
+ new HashMap<>(),
+ "");
+ InternalSchema internalSchema = schemaExtractor.toInternalSchema(paimonSchema);
+ List<InternalField> recordKeyFields =
+ primaryKeys.isEmpty()
+ ? Collections.emptyList()
+ : Collections.singletonList(expectedInternalField);
+ InternalSchema expectedSchema =
+ InternalSchema.builder()
+ .name("record")
+ .dataType(InternalType.RECORD)
+ .fields(Collections.singletonList(expectedInternalField))
+ .recordKeyFields(recordKeyFields)
+ .build();
+ assertEquals(expectedSchema, internalSchema);
+ }
+
+ @Test
+ void testCharField() {
+ DataField paimonField = new DataField(0, "char_field", new CharType(10));
+ InternalField expectedField =
+ InternalField.builder()
+ .name("char_field")
+ .fieldId(0)
+ .schema(
+ InternalSchema.builder()
+ .name("CHAR(10)")
+ .dataType(InternalType.STRING)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testVarcharField() {
+ DataField paimonField = new DataField(1, "varchar_field", new VarCharType(255));
+ InternalField expectedField =
+ InternalField.builder()
+ .name("varchar_field")
+ .fieldId(1)
+ .schema(
+ InternalSchema.builder()
+ .name("VARCHAR(255)")
+ .dataType(InternalType.STRING)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testBooleanField() {
+ DataField paimonField = new DataField(2, "boolean_field", new BooleanType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("boolean_field")
+ .fieldId(2)
+ .schema(
+ InternalSchema.builder()
+ .name("BOOLEAN")
+ .dataType(InternalType.BOOLEAN)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testTinyIntField() {
+ DataField paimonField = new DataField(3, "tinyint_field", new TinyIntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("tinyint_field")
+ .fieldId(3)
+ .schema(
+ InternalSchema.builder()
+ .name("TINYINT")
+ .dataType(InternalType.INT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testSmallIntField() {
+ DataField paimonField = new DataField(4, "smallint_field", new SmallIntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("smallint_field")
+ .fieldId(4)
+ .schema(
+ InternalSchema.builder()
+ .name("SMALLINT")
+ .dataType(InternalType.INT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testIntField() {
+ DataField paimonField = new DataField(5, "int_field", new IntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("int_field")
+ .fieldId(5)
+ .schema(
+ InternalSchema.builder()
+ .name("INT")
+ .dataType(InternalType.INT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testBigIntField() {
+ DataField paimonField = new DataField(6, "bigint_field", new BigIntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("bigint_field")
+ .fieldId(6)
+ .schema(
+ InternalSchema.builder()
+ .name("BIGINT")
+ .dataType(InternalType.LONG)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testFloatField() {
+ DataField paimonField = new DataField(7, "float_field", new FloatType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("float_field")
+ .fieldId(7)
+ .schema(
+ InternalSchema.builder()
+ .name("FLOAT")
+ .dataType(InternalType.FLOAT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testDoubleField() {
+ DataField paimonField = new DataField(8, "double_field", new DoubleType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("double_field")
+ .fieldId(8)
+ .schema(
+ InternalSchema.builder()
+ .name("DOUBLE")
+ .dataType(InternalType.DOUBLE)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testDateField() {
+ DataField paimonField = new DataField(9, "date_field", new DateType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("date_field")
+ .fieldId(9)
+ .schema(
+ InternalSchema.builder()
+ .name("DATE")
+ .dataType(InternalType.DATE)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testTimestampField() {
+ DataField paimonField = new DataField(10, "timestamp_field", new TimestampType(3));
Review Comment:
Can you add a test with millis precision as well if that is supported in Paimon?
##########
xtable-service/pom.xml:
##########
@@ -216,6 +216,17 @@
<scope>test</scope>
</dependency>
+ <!-- Paimon dependencies -->
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-bundle</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>paimon-spark-${spark.version.prefix}</artifactId>
+ <version>${paimon.version}</version>
Review Comment:
We typically define all the versions (even for test dependences) in the parent pom. Can you move this version declaration to the dependency management section there?
##########
xtable-service/src/test/java/org/apache/xtable/service/ITConversionService.java:
##########
@@ -97,6 +98,15 @@ public static void setupOnce() {
.sparkContext()
.hadoopConfiguration()
.set("parquet.avro.write-old-list-structure", "false");
+ sparkSession
+ .sparkContext()
+ .conf()
+ .set(
Review Comment:
Can these be set directly on the sparkConf on line 94?
--
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] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#issuecomment-3493864002
Thanks @mikedias!
--
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] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2484672849
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
+ private static final PaimonSchemaExtractor INSTANCE = new PaimonSchemaExtractor();
+
+ public static PaimonSchemaExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public InternalSchema toInternalSchema(TableSchema paimonSchema) {
+ RowType rowType = paimonSchema.logicalRowType();
+ List<InternalField> fields = toInternalFields(rowType);
+ return InternalSchema.builder()
+ .name("record")
+ .dataType(InternalType.RECORD)
+ .fields(fields)
+ .recordKeyFields(primaryKeyFields(paimonSchema, fields))
+ .build();
+ }
+
+ private List<InternalField> primaryKeyFields(
+ TableSchema paimonSchema, List<InternalField> internalFields) {
+ List<String> keys = paimonSchema.primaryKeys();
Review Comment:
Oh sorry, I missed this. No, primary keys and partition keys can't be nested in Paimon.
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2433965661
##########
xtable-core/src/test/java/org/apache/xtable/TestPaimonTable.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.manifest.BucketEntry;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.ParameterUtils;
+
+public class TestPaimonTable implements GenericTable<GenericRow, String> {
+
+ private final Random random = new Random();
+ private final FileStoreTable paimonTable;
+ private final String partitionField;
+
+ public TestPaimonTable(FileStoreTable paimonTable, String partitionField) {
+ this.paimonTable = paimonTable;
+ this.partitionField = partitionField;
+ }
+
+ public static GenericTable<GenericRow, String> createTable(
+ String tableName,
+ String partitionField,
+ Path tempDir,
+ Configuration hadoopConf,
+ boolean additionalColumns) {
+ String basePath = initBasePath(tempDir, tableName);
+ Catalog catalog = createFilesystemCatalog(basePath, hadoopConf);
+ FileStoreTable paimonTable = createTable(catalog, partitionField, additionalColumns);
+
+ System.out.println(
+ "Initialized Paimon test table at base path: "
+ + basePath
+ + " with partition field: "
+ + partitionField
+ + " and additional columns: "
+ + additionalColumns);
+
+ return new TestPaimonTable(paimonTable, partitionField);
+ }
+
+ public static Catalog createFilesystemCatalog(String basePath, Configuration hadoopConf) {
+ CatalogContext context = CatalogContext.create(new org.apache.paimon.fs.Path(basePath));
+ return CatalogFactory.createCatalog(context);
+ }
+
+ public static FileStoreTable createTable(
+ Catalog catalog, String partitionField, boolean additionalColumns) {
+ try {
+ catalog.createDatabase("test_db", true);
+ Identifier identifier = Identifier.create("test_db", "test_table");
+ Schema schema = buildSchema(partitionField, additionalColumns);
+ catalog.createTable(identifier, schema, true);
+ return (FileStoreTable) catalog.getTable(identifier);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static Schema buildSchema(String partitionField, boolean additionalColumns) {
+ Schema.Builder builder =
+ Schema.newBuilder()
+ .primaryKey("id")
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column("value", DataTypes.DOUBLE())
+ .column("created_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("updated_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("is_active", DataTypes.BOOLEAN())
+ .column("description", DataTypes.VARCHAR(255))
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .option("full-compaction.delta-commits", "1");
+
+ if (partitionField != null) {
+ builder
+ .primaryKey("id", partitionField)
+ .column(partitionField, DataTypes.STRING())
+ .partitionKeys(partitionField);
+ }
+
+ if (additionalColumns) {
+ builder.column("extra_info", DataTypes.STRING()).column("extra_value", DataTypes.DOUBLE());
+ }
+
+ return builder.build();
+ }
+
+ private GenericRow buildGenericRow(int rowIdx, TableSchema schema, String partitionValue) {
+ List<Object> rowValues = new ArrayList<>(schema.fields().size());
+ for (int i = 0; i < schema.fields().size(); i++) {
+ DataField field = schema.fields().get(i);
+ if (field.name().equals(partitionField)) {
+ rowValues.add(BinaryString.fromString(partitionValue));
+ } else if (field.type() instanceof IntType) {
+ rowValues.add(random.nextInt());
+ } else if (field.type() instanceof DoubleType) {
+ rowValues.add(random.nextDouble());
+ } else if (field.type() instanceof VarCharType) {
+ rowValues.add(BinaryString.fromString(field.name() + "_" + rowIdx + "_" + i));
+ } else if (field.type() instanceof LocalZonedTimestampType) {
+ rowValues.add(Timestamp.fromEpochMillis(System.currentTimeMillis()));
+ } else if (field.type() instanceof BooleanType) {
+ rowValues.add(random.nextBoolean());
+ } else {
+ throw new UnsupportedOperationException("Unsupported field type: " + field.type());
+ }
+ }
+
+ return GenericRow.of(rowValues.toArray());
+ }
+
+ private static String initBasePath(Path tempDir, String tableName) {
+ try {
+ Path basePath = tempDir.resolve(tableName);
+ Files.createDirectories(basePath);
+ return basePath.toUri().toString();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public List<GenericRow> insertRows(int numRows) {
+ String partitionValue = LEVEL_VALUES.get(0);
+ return insertRecordsToPartition(numRows, partitionValue);
+ }
+
+ @Override
+ public List<GenericRow> insertRecordsForSpecialPartition(int numRows) {
+ return insertRecordsToPartition(numRows, SPECIAL_PARTITION_VALUE);
+ }
+
+ private List<GenericRow> insertRecordsToPartition(int numRows, String partitionValue) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ List<GenericRow> rows = new ArrayList<>(numRows);
+ for (int i = 0; i < numRows; i++) {
+ GenericRow row = buildGenericRow(i, paimonTable.schema(), partitionValue);
+ writer.write(row);
+ rows.add(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ return rows;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to insert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void upsertRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to upsert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ row.setRowKind(RowKind.DELETE);
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete rows from Paimon table", e);
+ }
+ }
+
+ private void compactTable() {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ SnapshotReader snapshotReader = paimonTable.newSnapshotReader();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (BucketEntry bucketEntry : snapshotReader.bucketEntries()) {
+ writer.compact(bucketEntry.partition(), bucketEntry.bucket(), true);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to compact writes in Paimon table", e);
+ }
+ }
+
+ private static void commitWrites(BatchWriteBuilder batchWriteBuilder, BatchTableWrite writer)
+ throws Exception {
+ BatchTableCommit commit = batchWriteBuilder.newCommit();
+ List<CommitMessage> messages = writer.prepareCommit();
+ try {
+ commit.commit(messages);
+ } catch (Exception e) {
+ commit.abort(messages);
+ throw new RuntimeException("Failed to commit writes to Paimon table", e);
+ } finally {
+ commit.close();
+ }
+ }
+
+ @Override
+ public void deletePartition(String partitionValue) {
+ try (BatchTableCommit commit = paimonTable.newBatchWriteBuilder().newCommit()) {
+ commit.truncatePartitions(
+ ParameterUtils.getPartitions(partitionField + "=" + partitionValue));
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete partition from Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteSpecialPartition() {
+ deletePartition(SPECIAL_PARTITION_VALUE);
+ }
+
+ @Override
+ public String getBasePath() {
+ return paimonTable.location().toString();
+ }
+
+ @Override
+ public String getMetadataPath() {
+ return paimonTable.snapshotManager().snapshotDirectory().toString();
+ }
+
+ @Override
+ public String getOrderByColumn() {
+ return "id";
+ }
+
+ @Override
+ public void close() {}
+
+ @Override
+ public void reload() {}
+
+ @Override
+ public List<String> getColumnsToSelect() {
+ return paimonTable.schema().fieldNames().stream()
+ .filter(
+ // TODO Hudi thinks that paimon buckets are partition values, not sure how to handle it
+ // filtering out the partition field on the comparison for now
+ field -> !field.equals(partitionField))
Review Comment:
Okay... So, should we call out that Paimon -> Hudi won't work for partitioned values?
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2434209619
##########
xtable-core/src/test/java/org/apache/xtable/TestPaimonTable.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.manifest.BucketEntry;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.ParameterUtils;
+
+public class TestPaimonTable implements GenericTable<GenericRow, String> {
+
+ private final Random random = new Random();
+ private final FileStoreTable paimonTable;
+ private final String partitionField;
+
+ public TestPaimonTable(FileStoreTable paimonTable, String partitionField) {
+ this.paimonTable = paimonTable;
+ this.partitionField = partitionField;
+ }
+
+ public static GenericTable<GenericRow, String> createTable(
+ String tableName,
+ String partitionField,
+ Path tempDir,
+ Configuration hadoopConf,
+ boolean additionalColumns) {
+ String basePath = initBasePath(tempDir, tableName);
+ Catalog catalog = createFilesystemCatalog(basePath, hadoopConf);
+ FileStoreTable paimonTable = createTable(catalog, partitionField, additionalColumns);
+
+ System.out.println(
+ "Initialized Paimon test table at base path: "
+ + basePath
+ + " with partition field: "
+ + partitionField
+ + " and additional columns: "
+ + additionalColumns);
+
+ return new TestPaimonTable(paimonTable, partitionField);
+ }
+
+ public static Catalog createFilesystemCatalog(String basePath, Configuration hadoopConf) {
+ CatalogContext context = CatalogContext.create(new org.apache.paimon.fs.Path(basePath));
+ return CatalogFactory.createCatalog(context);
+ }
+
+ public static FileStoreTable createTable(
+ Catalog catalog, String partitionField, boolean additionalColumns) {
+ try {
+ catalog.createDatabase("test_db", true);
+ Identifier identifier = Identifier.create("test_db", "test_table");
+ Schema schema = buildSchema(partitionField, additionalColumns);
+ catalog.createTable(identifier, schema, true);
+ return (FileStoreTable) catalog.getTable(identifier);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static Schema buildSchema(String partitionField, boolean additionalColumns) {
+ Schema.Builder builder =
+ Schema.newBuilder()
+ .primaryKey("id")
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column("value", DataTypes.DOUBLE())
+ .column("created_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("updated_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("is_active", DataTypes.BOOLEAN())
+ .column("description", DataTypes.VARCHAR(255))
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .option("full-compaction.delta-commits", "1");
+
+ if (partitionField != null) {
+ builder
+ .primaryKey("id", partitionField)
+ .column(partitionField, DataTypes.STRING())
+ .partitionKeys(partitionField);
+ }
+
+ if (additionalColumns) {
+ builder.column("extra_info", DataTypes.STRING()).column("extra_value", DataTypes.DOUBLE());
+ }
+
+ return builder.build();
+ }
+
+ private GenericRow buildGenericRow(int rowIdx, TableSchema schema, String partitionValue) {
+ List<Object> rowValues = new ArrayList<>(schema.fields().size());
+ for (int i = 0; i < schema.fields().size(); i++) {
+ DataField field = schema.fields().get(i);
+ if (field.name().equals(partitionField)) {
+ rowValues.add(BinaryString.fromString(partitionValue));
+ } else if (field.type() instanceof IntType) {
+ rowValues.add(random.nextInt());
+ } else if (field.type() instanceof DoubleType) {
+ rowValues.add(random.nextDouble());
+ } else if (field.type() instanceof VarCharType) {
+ rowValues.add(BinaryString.fromString(field.name() + "_" + rowIdx + "_" + i));
+ } else if (field.type() instanceof LocalZonedTimestampType) {
+ rowValues.add(Timestamp.fromEpochMillis(System.currentTimeMillis()));
+ } else if (field.type() instanceof BooleanType) {
+ rowValues.add(random.nextBoolean());
+ } else {
+ throw new UnsupportedOperationException("Unsupported field type: " + field.type());
+ }
+ }
+
+ return GenericRow.of(rowValues.toArray());
+ }
+
+ private static String initBasePath(Path tempDir, String tableName) {
+ try {
+ Path basePath = tempDir.resolve(tableName);
+ Files.createDirectories(basePath);
+ return basePath.toUri().toString();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public List<GenericRow> insertRows(int numRows) {
+ String partitionValue = LEVEL_VALUES.get(0);
+ return insertRecordsToPartition(numRows, partitionValue);
+ }
+
+ @Override
+ public List<GenericRow> insertRecordsForSpecialPartition(int numRows) {
+ return insertRecordsToPartition(numRows, SPECIAL_PARTITION_VALUE);
+ }
+
+ private List<GenericRow> insertRecordsToPartition(int numRows, String partitionValue) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ List<GenericRow> rows = new ArrayList<>(numRows);
+ for (int i = 0; i < numRows; i++) {
+ GenericRow row = buildGenericRow(i, paimonTable.schema(), partitionValue);
+ writer.write(row);
+ rows.add(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ return rows;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to insert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void upsertRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to upsert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ row.setRowKind(RowKind.DELETE);
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete rows from Paimon table", e);
+ }
+ }
+
+ private void compactTable() {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ SnapshotReader snapshotReader = paimonTable.newSnapshotReader();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (BucketEntry bucketEntry : snapshotReader.bucketEntries()) {
+ writer.compact(bucketEntry.partition(), bucketEntry.bucket(), true);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to compact writes in Paimon table", e);
+ }
+ }
+
+ private static void commitWrites(BatchWriteBuilder batchWriteBuilder, BatchTableWrite writer)
+ throws Exception {
+ BatchTableCommit commit = batchWriteBuilder.newCommit();
+ List<CommitMessage> messages = writer.prepareCommit();
+ try {
+ commit.commit(messages);
+ } catch (Exception e) {
+ commit.abort(messages);
+ throw new RuntimeException("Failed to commit writes to Paimon table", e);
+ } finally {
+ commit.close();
+ }
+ }
+
+ @Override
+ public void deletePartition(String partitionValue) {
+ try (BatchTableCommit commit = paimonTable.newBatchWriteBuilder().newCommit()) {
+ commit.truncatePartitions(
+ ParameterUtils.getPartitions(partitionField + "=" + partitionValue));
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete partition from Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteSpecialPartition() {
+ deletePartition(SPECIAL_PARTITION_VALUE);
+ }
+
+ @Override
+ public String getBasePath() {
+ return paimonTable.location().toString();
+ }
+
+ @Override
+ public String getMetadataPath() {
+ return paimonTable.snapshotManager().snapshotDirectory().toString();
+ }
+
+ @Override
+ public String getOrderByColumn() {
+ return "id";
+ }
+
+ @Override
+ public void close() {}
+
+ @Override
+ public void reload() {}
+
+ @Override
+ public List<String> getColumnsToSelect() {
+ return paimonTable.schema().fieldNames().stream()
+ .filter(
+ // TODO Hudi thinks that paimon buckets are partition values, not sure how to handle it
+ // filtering out the partition field on the comparison for now
+ field -> !field.equals(partitionField))
Review Comment:
Sounds good, I'll keep excluding the partition column here.
I'll finish the unit tests and move the PR out of draft :)
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2336863920
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
There is a way to read by setting a default mapping. Not all query engines will use this but spark does.
I think we have different versions of `parquet-tools`. I will run the test and see what I find.
--
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] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2484372778
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonConversionSource.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.paimon;
+
+import static org.apache.xtable.model.storage.DataLayoutStrategy.HIVE_STYLE_PARTITION;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.List;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.xtable.exception.ReadException;
+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.DataLayoutStrategy;
+import org.apache.xtable.model.storage.InternalDataFile;
+import org.apache.xtable.model.storage.PartitionFileGroup;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Log4j2
+public class PaimonConversionSource implements ConversionSource<Snapshot> {
+
+ private final FileStoreTable paimonTable;
+ private final SchemaManager schemaManager;
+ private final SnapshotManager snapshotManager;
+
+ private final PaimonDataFileExtractor dataFileExtractor = PaimonDataFileExtractor.getInstance();
+ private final PaimonSchemaExtractor schemaExtractor = PaimonSchemaExtractor.getInstance();
+ private final PaimonPartitionExtractor partitionSpecExtractor =
+ PaimonPartitionExtractor.getInstance();
+
+ public PaimonConversionSource(FileStoreTable paimonTable) {
+ this.paimonTable = paimonTable;
+ this.schemaManager = paimonTable.schemaManager();
+ this.snapshotManager = paimonTable.snapshotManager();
+ }
+
+ @Override
+ public InternalTable getTable(Snapshot snapshot) {
+ TableSchema paimonSchema = schemaManager.schema(snapshot.schemaId());
+ InternalSchema internalSchema = schemaExtractor.toInternalSchema(paimonSchema);
+
+ List<String> partitionKeys = paimonTable.partitionKeys();
+ List<InternalPartitionField> partitioningFields =
+ partitionSpecExtractor.toInternalPartitionFields(partitionKeys, internalSchema);
+
+ DataLayoutStrategy dataLayoutStrategy =
+ partitioningFields.isEmpty() ? DataLayoutStrategy.FLAT : HIVE_STYLE_PARTITION;
Review Comment:
[It is always Hive Style](https://paimon.apache.org/docs/1.0/concepts/basic-concepts/#partition) (with the field names)
Probably best to always use `DataLayoutStrategy.HIVE_STYLE_PARTITION` even if there are no partitions. I'll change that.
--
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] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2484413662
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonDataFileExtractor.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.paimon;
+
+import java.util.*;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+
+import org.apache.xtable.model.stat.ColumnStat;
+import org.apache.xtable.model.storage.InternalDataFile;
+
+public class PaimonDataFileExtractor {
+
+ private final PaimonPartitionExtractor partitionExtractor =
+ PaimonPartitionExtractor.getInstance();
+
+ private static final PaimonDataFileExtractor INSTANCE = new PaimonDataFileExtractor();
+
+ public static PaimonDataFileExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalDataFile> toInternalDataFiles(FileStoreTable table, Snapshot snapshot) {
+ List<InternalDataFile> result = new ArrayList<>();
+ Iterator<ManifestEntry> manifestEntryIterator =
+ newSnapshotReader(table, snapshot).readFileIterator();
+ while (manifestEntryIterator.hasNext()) {
+ result.add(toInternalDataFile(table, manifestEntryIterator.next()));
+ }
+ return result;
+ }
+
+ private InternalDataFile toInternalDataFile(FileStoreTable table, ManifestEntry entry) {
+ return InternalDataFile.builder()
+ .physicalPath(toFullPhysicalPath(table, entry))
+ .fileSizeBytes(entry.file().fileSize())
+ .lastModified(entry.file().creationTimeEpochMillis())
+ .recordCount(entry.file().rowCount())
+ .partitionValues(partitionExtractor.toPartitionValues(table, entry.partition()))
+ .columnStats(toColumnStats(entry.file()))
+ .build();
+ }
+
+ private String toFullPhysicalPath(FileStoreTable table, ManifestEntry entry) {
+ String basePath = table.location().toString();
+ String bucketPath = "bucket-" + entry.bucket();
+ String filePath = entry.file().fileName();
+
+ Optional<String> partitionPath = partitionExtractor.toPartitionPath(table, entry.partition());
+ if (partitionPath.isPresent()) {
+ return String.join("/", basePath, partitionPath.get(), bucketPath, filePath);
+ } else {
+ return String.join("/", basePath, bucketPath, filePath);
+ }
+ }
+
+ private List<ColumnStat> toColumnStats(DataFileMeta file) {
+ // TODO: Implement logic to extract column stats from the file meta
Review Comment:
https://github.com/apache/incubator-xtable/issues/755
--
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] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#issuecomment-3477805046
Thanks for the awesome review @the-other-tim-brown! I've made the changes in the latest [commit](https://github.com/apache/incubator-xtable/pull/742/commits/fc72c381d61f46f32b429111727ceaa3b971e860)
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2412266792
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
I have found a way to get past this issue, working on a PR now for it: https://github.com/apache/incubator-xtable/pull/749
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2331955040
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonConversionSourceProvider.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.paimon;
+
+import java.io.IOException;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+public class PaimonConversionSourceProvider extends ConversionSourceProvider<Snapshot> {
+ @Override
+ public ConversionSource<Snapshot> getConversionSourceInstance(SourceTable sourceTableConfig) {
+ try {
+ Options catalogOptions = new Options();
+ CatalogContext context = CatalogContext.create(catalogOptions, hadoopConf);
+
+ Path path = new Path(sourceTableConfig.getDataPath());
+ FileIO fileIO = FileIO.get(path, context);
+ FileStoreTable paimonTable = FileStoreTableFactory.create(fileIO, path);
+
+ return new PaimonConversionSource(paimonTable);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
Review Comment:
got it, will change!
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "rahil-c (via GitHub)" <gi...@apache.org>.
rahil-c commented on PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#issuecomment-3264718773
Thanks @mikedias for your contribution!
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2336921137
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
I see the following for the schema:
```
message paimon_schema {
required int32 _KEY_id = 1073741823;
required int64 _SEQUENCE_NUMBER = 2147483646;
required int32 _VALUE_KIND (INTEGER(8,true)) = 2147483645;
required int32 id = 0;
optional binary name (STRING) = 1;
optional double value = 2;
optional int64 created_at (TIMESTAMP(MICROS,true)) = 3;
optional int64 updated_at (TIMESTAMP(MICROS,true)) = 4;
optional boolean is_active = 5;
optional binary description (STRING) = 6;
required binary level (STRING) = 7;
}
```
The first 3 fields do not have a field index set somehow.
--
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] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2488430331
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonPartitionExtractor.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.paimon;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
+
+import org.apache.xtable.model.schema.InternalField;
+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;
+
+/** Extracts partition spec for Paimon as identity transforms on partition keys. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonPartitionExtractor {
+
+ private final PaimonSchemaExtractor paimonSchemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private static final PaimonPartitionExtractor INSTANCE = new PaimonPartitionExtractor();
+
+ public static PaimonPartitionExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalPartitionField> toInternalPartitionFields(
+ List<String> partitionKeys, InternalSchema schema) {
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return partitionKeys.stream()
+ .map(key -> toPartitionField(key, schema))
+ .collect(Collectors.toList());
+ }
+
+ public List<PartitionValue> toPartitionValues(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ InternalSchema internalSchema = paimonSchemaExtractor.toInternalSchema(table.schema());
+
+ List<PartitionValue> partitionValues = new ArrayList<>();
+ for (Map.Entry<String, String> entry :
+ partitionComputer.generatePartValues(partition).entrySet()) {
+ PartitionValue partitionValue =
+ PartitionValue.builder()
+ .partitionField(toPartitionField(entry.getKey(), internalSchema))
+ .range(Range.scalar(entry.getValue()))
+ .build();
+ partitionValues.add(partitionValue);
+ }
+ return partitionValues;
+ }
+
+ public Optional<String> toPartitionPath(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ return partitionComputer.generatePartValues(partition).entrySet().stream()
+ .map(e -> e.getKey() + "=" + e.getValue())
+ .reduce((a, b) -> a + "/" + b);
+ }
+
+ private InternalPartitionField toPartitionField(String key, InternalSchema schema) {
+ InternalField sourceField =
+ findField(schema, key)
+ .orElseThrow(
+ () -> new IllegalArgumentException("Partition key not found in schema: " + key));
+ return InternalPartitionField.builder()
+ .sourceField(sourceField)
+ .transformType(PartitionTransformType.VALUE)
Review Comment:
okay, probably should as a follow-up task.
I'm curious if that partition transformation type is specific to Iceberg or if Hudi or Delta also have a similar concept. If that is specific to Iceberg, perhaps the logic behind parsing partition ranges should be handled only in the Iceberg target code?
--
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] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown merged PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2425005237
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
@the-other-tim-brown I've rebased it, and the Iceberg tests are passing now! Thank you so much for finding the root cause and fixing 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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2433159130
##########
xtable-core/src/test/java/org/apache/xtable/TestPaimonTable.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.manifest.BucketEntry;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.ParameterUtils;
+
+public class TestPaimonTable implements GenericTable<GenericRow, String> {
+
+ private final Random random = new Random();
+ private final FileStoreTable paimonTable;
+ private final String partitionField;
+
+ public TestPaimonTable(FileStoreTable paimonTable, String partitionField) {
+ this.paimonTable = paimonTable;
+ this.partitionField = partitionField;
+ }
+
+ public static GenericTable<GenericRow, String> createTable(
+ String tableName,
+ String partitionField,
+ Path tempDir,
+ Configuration hadoopConf,
+ boolean additionalColumns) {
+ String basePath = initBasePath(tempDir, tableName);
+ Catalog catalog = createFilesystemCatalog(basePath, hadoopConf);
+ FileStoreTable paimonTable = createTable(catalog, partitionField, additionalColumns);
+
+ System.out.println(
+ "Initialized Paimon test table at base path: "
+ + basePath
+ + " with partition field: "
+ + partitionField
+ + " and additional columns: "
+ + additionalColumns);
+
+ return new TestPaimonTable(paimonTable, partitionField);
+ }
+
+ public static Catalog createFilesystemCatalog(String basePath, Configuration hadoopConf) {
+ CatalogContext context = CatalogContext.create(new org.apache.paimon.fs.Path(basePath));
+ return CatalogFactory.createCatalog(context);
+ }
+
+ public static FileStoreTable createTable(
+ Catalog catalog, String partitionField, boolean additionalColumns) {
+ try {
+ catalog.createDatabase("test_db", true);
+ Identifier identifier = Identifier.create("test_db", "test_table");
+ Schema schema = buildSchema(partitionField, additionalColumns);
+ catalog.createTable(identifier, schema, true);
+ return (FileStoreTable) catalog.getTable(identifier);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static Schema buildSchema(String partitionField, boolean additionalColumns) {
+ Schema.Builder builder =
+ Schema.newBuilder()
+ .primaryKey("id")
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column("value", DataTypes.DOUBLE())
+ .column("created_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("updated_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("is_active", DataTypes.BOOLEAN())
+ .column("description", DataTypes.VARCHAR(255))
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .option("full-compaction.delta-commits", "1");
+
+ if (partitionField != null) {
+ builder
+ .primaryKey("id", partitionField)
+ .column(partitionField, DataTypes.STRING())
+ .partitionKeys(partitionField);
+ }
+
+ if (additionalColumns) {
+ builder.column("extra_info", DataTypes.STRING()).column("extra_value", DataTypes.DOUBLE());
+ }
+
+ return builder.build();
+ }
+
+ private GenericRow buildGenericRow(int rowIdx, TableSchema schema, String partitionValue) {
+ List<Object> rowValues = new ArrayList<>(schema.fields().size());
+ for (int i = 0; i < schema.fields().size(); i++) {
+ DataField field = schema.fields().get(i);
+ if (field.name().equals(partitionField)) {
+ rowValues.add(BinaryString.fromString(partitionValue));
+ } else if (field.type() instanceof IntType) {
+ rowValues.add(random.nextInt());
+ } else if (field.type() instanceof DoubleType) {
+ rowValues.add(random.nextDouble());
+ } else if (field.type() instanceof VarCharType) {
+ rowValues.add(BinaryString.fromString(field.name() + "_" + rowIdx + "_" + i));
+ } else if (field.type() instanceof LocalZonedTimestampType) {
+ rowValues.add(Timestamp.fromEpochMillis(System.currentTimeMillis()));
+ } else if (field.type() instanceof BooleanType) {
+ rowValues.add(random.nextBoolean());
+ } else {
+ throw new UnsupportedOperationException("Unsupported field type: " + field.type());
+ }
+ }
+
+ return GenericRow.of(rowValues.toArray());
+ }
+
+ private static String initBasePath(Path tempDir, String tableName) {
+ try {
+ Path basePath = tempDir.resolve(tableName);
+ Files.createDirectories(basePath);
+ return basePath.toUri().toString();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public List<GenericRow> insertRows(int numRows) {
+ String partitionValue = LEVEL_VALUES.get(0);
+ return insertRecordsToPartition(numRows, partitionValue);
+ }
+
+ @Override
+ public List<GenericRow> insertRecordsForSpecialPartition(int numRows) {
+ return insertRecordsToPartition(numRows, SPECIAL_PARTITION_VALUE);
+ }
+
+ private List<GenericRow> insertRecordsToPartition(int numRows, String partitionValue) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ List<GenericRow> rows = new ArrayList<>(numRows);
+ for (int i = 0; i < numRows; i++) {
+ GenericRow row = buildGenericRow(i, paimonTable.schema(), partitionValue);
+ writer.write(row);
+ rows.add(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ return rows;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to insert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void upsertRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to upsert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ row.setRowKind(RowKind.DELETE);
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete rows from Paimon table", e);
+ }
+ }
+
+ private void compactTable() {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ SnapshotReader snapshotReader = paimonTable.newSnapshotReader();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (BucketEntry bucketEntry : snapshotReader.bucketEntries()) {
+ writer.compact(bucketEntry.partition(), bucketEntry.bucket(), true);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to compact writes in Paimon table", e);
+ }
+ }
+
+ private static void commitWrites(BatchWriteBuilder batchWriteBuilder, BatchTableWrite writer)
+ throws Exception {
+ BatchTableCommit commit = batchWriteBuilder.newCommit();
+ List<CommitMessage> messages = writer.prepareCommit();
+ try {
+ commit.commit(messages);
+ } catch (Exception e) {
+ commit.abort(messages);
+ throw new RuntimeException("Failed to commit writes to Paimon table", e);
+ } finally {
+ commit.close();
+ }
+ }
+
+ @Override
+ public void deletePartition(String partitionValue) {
+ try (BatchTableCommit commit = paimonTable.newBatchWriteBuilder().newCommit()) {
+ commit.truncatePartitions(
+ ParameterUtils.getPartitions(partitionField + "=" + partitionValue));
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete partition from Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteSpecialPartition() {
+ deletePartition(SPECIAL_PARTITION_VALUE);
+ }
+
+ @Override
+ public String getBasePath() {
+ return paimonTable.location().toString();
+ }
+
+ @Override
+ public String getMetadataPath() {
+ return paimonTable.snapshotManager().snapshotDirectory().toString();
+ }
+
+ @Override
+ public String getOrderByColumn() {
+ return "id";
+ }
+
+ @Override
+ public void close() {}
+
+ @Override
+ public void reload() {}
+
+ @Override
+ public List<String> getColumnsToSelect() {
+ return paimonTable.schema().fieldNames().stream()
+ .filter(
+ // TODO Hudi thinks that paimon buckets are partition values, not sure how to handle it
+ // filtering out the partition field on the comparison for now
+ field -> !field.equals(partitionField))
Review Comment:
This seems to be a deeper problem within Hudi. Even if the partition is properly extracted on the XTable side, the Hudi reader will not be able to understand the structure.
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2336960824
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
I think the issue comes from the starting index. When we sync to Iceberg, the library will go through a step [TypeUtil.assignFreshIds](https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/TableMetadata.java#L120). This will shift all the IDs to start at 1 instead of 0 causing the mismatch in column types.
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2412089891
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
It seems to be on the [ParquetSchemaConverter](https://github.com/apache/paimon/blob/master/paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetSchemaConverter.java), where it uses the DataField ids defined on the links above
--
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] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2484391853
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonConversionSource.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.paimon;
+
+import static org.apache.xtable.model.storage.DataLayoutStrategy.HIVE_STYLE_PARTITION;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.List;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.xtable.exception.ReadException;
+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.DataLayoutStrategy;
+import org.apache.xtable.model.storage.InternalDataFile;
+import org.apache.xtable.model.storage.PartitionFileGroup;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+@Log4j2
+public class PaimonConversionSource implements ConversionSource<Snapshot> {
+
+ private final FileStoreTable paimonTable;
+ private final SchemaManager schemaManager;
+ private final SnapshotManager snapshotManager;
+
+ private final PaimonDataFileExtractor dataFileExtractor = PaimonDataFileExtractor.getInstance();
+ private final PaimonSchemaExtractor schemaExtractor = PaimonSchemaExtractor.getInstance();
+ private final PaimonPartitionExtractor partitionSpecExtractor =
+ PaimonPartitionExtractor.getInstance();
+
+ public PaimonConversionSource(FileStoreTable paimonTable) {
+ this.paimonTable = paimonTable;
+ this.schemaManager = paimonTable.schemaManager();
+ this.snapshotManager = paimonTable.snapshotManager();
+ }
+
+ @Override
+ public InternalTable getTable(Snapshot snapshot) {
+ TableSchema paimonSchema = schemaManager.schema(snapshot.schemaId());
+ InternalSchema internalSchema = schemaExtractor.toInternalSchema(paimonSchema);
+
+ List<String> partitionKeys = paimonTable.partitionKeys();
+ List<InternalPartitionField> partitioningFields =
+ partitionSpecExtractor.toInternalPartitionFields(partitionKeys, internalSchema);
+
+ DataLayoutStrategy dataLayoutStrategy =
+ partitioningFields.isEmpty() ? DataLayoutStrategy.FLAT : HIVE_STYLE_PARTITION;
+
+ return InternalTable.builder()
+ .name(paimonTable.name())
+ .tableFormat(TableFormat.PAIMON)
+ .readSchema(internalSchema)
+ .layoutStrategy(dataLayoutStrategy)
+ .basePath(paimonTable.location().toString())
+ .partitioningFields(partitioningFields)
+ .latestCommitTime(Instant.ofEpochMilli(snapshot.timeMillis()))
+ .latestMetadataPath(snapshotManager.snapshotPath(snapshot.id()).toString())
+ .build();
+ }
+
+ @Override
+ public InternalTable getCurrentTable() {
+ SnapshotManager snapshotManager = paimonTable.snapshotManager();
+ Snapshot snapshot = snapshotManager.latestSnapshot();
+ if (snapshot == null) {
+ throw new ReadException("No snapshots found for table " + paimonTable.name());
+ }
+ return getTable(snapshot);
+ }
+
+ @Override
+ public InternalSnapshot getCurrentSnapshot() {
+ SnapshotManager snapshotManager = paimonTable.snapshotManager();
+ Snapshot snapshot = snapshotManager.latestSnapshot();
+ if (snapshot == null) {
+ throw new ReadException("No snapshots found for table " + paimonTable.name());
+ }
+
+ InternalTable internalTable = getTable(snapshot);
+ List<InternalDataFile> internalDataFiles =
+ dataFileExtractor.toInternalDataFiles(paimonTable, snapshot);
+
+ return InternalSnapshot.builder()
+ .table(internalTable)
+ .version(Long.toString(snapshot.timeMillis()))
+ .partitionedDataFiles(PartitionFileGroup.fromFiles(internalDataFiles))
+ // TODO : Implement pending commits extraction, required for incremental sync
Review Comment:
https://github.com/apache/incubator-xtable/issues/754
--
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] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2484403552
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonConversionSourceProvider.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.paimon;
+
+import java.io.IOException;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.exception.ReadException;
+import org.apache.xtable.spi.extractor.ConversionSource;
+
+public class PaimonConversionSourceProvider extends ConversionSourceProvider<Snapshot> {
+ @Override
+ public ConversionSource<Snapshot> getConversionSourceInstance(SourceTable sourceTableConfig) {
+ try {
+ Options catalogOptions = new Options();
Review Comment:
Probably yes, but I'm not sure what they look like right now...
--
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] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2483769170
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
+ private static final PaimonSchemaExtractor INSTANCE = new PaimonSchemaExtractor();
+
+ public static PaimonSchemaExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public InternalSchema toInternalSchema(TableSchema paimonSchema) {
+ RowType rowType = paimonSchema.logicalRowType();
+ List<InternalField> fields = toInternalFields(rowType);
+ return InternalSchema.builder()
+ .name("record")
+ .dataType(InternalType.RECORD)
+ .fields(fields)
+ .recordKeyFields(primaryKeyFields(paimonSchema, fields))
+ .build();
+ }
+
+ private List<InternalField> primaryKeyFields(
+ TableSchema paimonSchema, List<InternalField> internalFields) {
+ List<String> keys = paimonSchema.primaryKeys();
Review Comment:
@mikedias any updates on 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] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2488430331
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonPartitionExtractor.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.paimon;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
+
+import org.apache.xtable.model.schema.InternalField;
+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;
+
+/** Extracts partition spec for Paimon as identity transforms on partition keys. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonPartitionExtractor {
+
+ private final PaimonSchemaExtractor paimonSchemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private static final PaimonPartitionExtractor INSTANCE = new PaimonPartitionExtractor();
+
+ public static PaimonPartitionExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalPartitionField> toInternalPartitionFields(
+ List<String> partitionKeys, InternalSchema schema) {
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return partitionKeys.stream()
+ .map(key -> toPartitionField(key, schema))
+ .collect(Collectors.toList());
+ }
+
+ public List<PartitionValue> toPartitionValues(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ InternalSchema internalSchema = paimonSchemaExtractor.toInternalSchema(table.schema());
+
+ List<PartitionValue> partitionValues = new ArrayList<>();
+ for (Map.Entry<String, String> entry :
+ partitionComputer.generatePartValues(partition).entrySet()) {
+ PartitionValue partitionValue =
+ PartitionValue.builder()
+ .partitionField(toPartitionField(entry.getKey(), internalSchema))
+ .range(Range.scalar(entry.getValue()))
+ .build();
+ partitionValues.add(partitionValue);
+ }
+ return partitionValues;
+ }
+
+ public Optional<String> toPartitionPath(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ return partitionComputer.generatePartValues(partition).entrySet().stream()
+ .map(e -> e.getKey() + "=" + e.getValue())
+ .reduce((a, b) -> a + "/" + b);
+ }
+
+ private InternalPartitionField toPartitionField(String key, InternalSchema schema) {
+ InternalField sourceField =
+ findField(schema, key)
+ .orElseThrow(
+ () -> new IllegalArgumentException("Partition key not found in schema: " + key));
+ return InternalPartitionField.builder()
+ .sourceField(sourceField)
+ .transformType(PartitionTransformType.VALUE)
Review Comment:
okay, probably should implement as a follow-up task, if that's ok.
I'm curious if that partition transformation type is specific to Iceberg or if Hudi or Delta also have a similar concept. If that is specific to Iceberg, perhaps the logic behind parsing partition ranges should be handled only in the Iceberg target code?
--
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] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#issuecomment-3448996047
@the-other-tim-brown I've added the missing unit tests, and I think it's good to review. Let me know if there is something I'm missing or you want to approach differently.
--
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] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2486686539
##########
xtable-core/src/test/java/org/apache/xtable/paimon/TestPaimonSchemaExtractor.java:
##########
@@ -0,0 +1,538 @@
+/*
+ * 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.paimon;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.CharType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DateType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.FloatType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.SmallIntType;
+import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.types.TinyIntType;
+import org.apache.paimon.types.VarCharType;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+
+public class TestPaimonSchemaExtractor {
+ private static final PaimonSchemaExtractor schemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private void assertField(DataField paimonField, InternalField expectedInternalField) {
+ assertField(paimonField, expectedInternalField, Collections.emptyList());
+ }
+
+ private void assertField(
+ DataField paimonField, InternalField expectedInternalField, List<String> primaryKeys) {
+ TableSchema paimonSchema =
+ new TableSchema(
+ 0,
+ Collections.singletonList(paimonField),
+ 0,
+ Collections.emptyList(),
+ primaryKeys,
+ new HashMap<>(),
+ "");
+ InternalSchema internalSchema = schemaExtractor.toInternalSchema(paimonSchema);
+ List<InternalField> recordKeyFields =
+ primaryKeys.isEmpty()
+ ? Collections.emptyList()
+ : Collections.singletonList(expectedInternalField);
+ InternalSchema expectedSchema =
+ InternalSchema.builder()
+ .name("record")
+ .dataType(InternalType.RECORD)
+ .fields(Collections.singletonList(expectedInternalField))
+ .recordKeyFields(recordKeyFields)
+ .build();
+ assertEquals(expectedSchema, internalSchema);
+ }
+
+ @Test
+ void testCharField() {
+ DataField paimonField = new DataField(0, "char_field", new CharType(10));
+ InternalField expectedField =
+ InternalField.builder()
+ .name("char_field")
+ .fieldId(0)
+ .schema(
+ InternalSchema.builder()
+ .name("CHAR(10)")
+ .dataType(InternalType.STRING)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testVarcharField() {
+ DataField paimonField = new DataField(1, "varchar_field", new VarCharType(255));
+ InternalField expectedField =
+ InternalField.builder()
+ .name("varchar_field")
+ .fieldId(1)
+ .schema(
+ InternalSchema.builder()
+ .name("VARCHAR(255)")
+ .dataType(InternalType.STRING)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testBooleanField() {
+ DataField paimonField = new DataField(2, "boolean_field", new BooleanType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("boolean_field")
+ .fieldId(2)
+ .schema(
+ InternalSchema.builder()
+ .name("BOOLEAN")
+ .dataType(InternalType.BOOLEAN)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testTinyIntField() {
+ DataField paimonField = new DataField(3, "tinyint_field", new TinyIntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("tinyint_field")
+ .fieldId(3)
+ .schema(
+ InternalSchema.builder()
+ .name("TINYINT")
+ .dataType(InternalType.INT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testSmallIntField() {
+ DataField paimonField = new DataField(4, "smallint_field", new SmallIntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("smallint_field")
+ .fieldId(4)
+ .schema(
+ InternalSchema.builder()
+ .name("SMALLINT")
+ .dataType(InternalType.INT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testIntField() {
+ DataField paimonField = new DataField(5, "int_field", new IntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("int_field")
+ .fieldId(5)
+ .schema(
+ InternalSchema.builder()
+ .name("INT")
+ .dataType(InternalType.INT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testBigIntField() {
+ DataField paimonField = new DataField(6, "bigint_field", new BigIntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("bigint_field")
+ .fieldId(6)
+ .schema(
+ InternalSchema.builder()
+ .name("BIGINT")
+ .dataType(InternalType.LONG)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testFloatField() {
+ DataField paimonField = new DataField(7, "float_field", new FloatType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("float_field")
+ .fieldId(7)
+ .schema(
+ InternalSchema.builder()
+ .name("FLOAT")
+ .dataType(InternalType.FLOAT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testDoubleField() {
+ DataField paimonField = new DataField(8, "double_field", new DoubleType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("double_field")
+ .fieldId(8)
+ .schema(
+ InternalSchema.builder()
+ .name("DOUBLE")
+ .dataType(InternalType.DOUBLE)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testDateField() {
+ DataField paimonField = new DataField(9, "date_field", new DateType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("date_field")
+ .fieldId(9)
+ .schema(
+ InternalSchema.builder()
+ .name("DATE")
+ .dataType(InternalType.DATE)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9})
+ void testTimestampField(int precision) {
+ DataField paimonField = new DataField(10, "timestamp_field", new TimestampType(precision));
+ Map<InternalSchema.MetadataKey, Object> timestampMetadata =
+ Collections.singletonMap(
+ InternalSchema.MetadataKey.TIMESTAMP_PRECISION, InternalSchema.MetadataValue.MICROS);
Review Comment:
This looks like a bug based on my understanding of the docs. It should be not always be micros
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2434082130
##########
xtable-core/src/test/java/org/apache/xtable/TestPaimonTable.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.manifest.BucketEntry;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.ParameterUtils;
+
+public class TestPaimonTable implements GenericTable<GenericRow, String> {
+
+ private final Random random = new Random();
+ private final FileStoreTable paimonTable;
+ private final String partitionField;
+
+ public TestPaimonTable(FileStoreTable paimonTable, String partitionField) {
+ this.paimonTable = paimonTable;
+ this.partitionField = partitionField;
+ }
+
+ public static GenericTable<GenericRow, String> createTable(
+ String tableName,
+ String partitionField,
+ Path tempDir,
+ Configuration hadoopConf,
+ boolean additionalColumns) {
+ String basePath = initBasePath(tempDir, tableName);
+ Catalog catalog = createFilesystemCatalog(basePath, hadoopConf);
+ FileStoreTable paimonTable = createTable(catalog, partitionField, additionalColumns);
+
+ System.out.println(
+ "Initialized Paimon test table at base path: "
+ + basePath
+ + " with partition field: "
+ + partitionField
+ + " and additional columns: "
+ + additionalColumns);
+
+ return new TestPaimonTable(paimonTable, partitionField);
+ }
+
+ public static Catalog createFilesystemCatalog(String basePath, Configuration hadoopConf) {
+ CatalogContext context = CatalogContext.create(new org.apache.paimon.fs.Path(basePath));
+ return CatalogFactory.createCatalog(context);
+ }
+
+ public static FileStoreTable createTable(
+ Catalog catalog, String partitionField, boolean additionalColumns) {
+ try {
+ catalog.createDatabase("test_db", true);
+ Identifier identifier = Identifier.create("test_db", "test_table");
+ Schema schema = buildSchema(partitionField, additionalColumns);
+ catalog.createTable(identifier, schema, true);
+ return (FileStoreTable) catalog.getTable(identifier);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static Schema buildSchema(String partitionField, boolean additionalColumns) {
+ Schema.Builder builder =
+ Schema.newBuilder()
+ .primaryKey("id")
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column("value", DataTypes.DOUBLE())
+ .column("created_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("updated_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("is_active", DataTypes.BOOLEAN())
+ .column("description", DataTypes.VARCHAR(255))
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .option("full-compaction.delta-commits", "1");
+
+ if (partitionField != null) {
+ builder
+ .primaryKey("id", partitionField)
+ .column(partitionField, DataTypes.STRING())
+ .partitionKeys(partitionField);
+ }
+
+ if (additionalColumns) {
+ builder.column("extra_info", DataTypes.STRING()).column("extra_value", DataTypes.DOUBLE());
+ }
+
+ return builder.build();
+ }
+
+ private GenericRow buildGenericRow(int rowIdx, TableSchema schema, String partitionValue) {
+ List<Object> rowValues = new ArrayList<>(schema.fields().size());
+ for (int i = 0; i < schema.fields().size(); i++) {
+ DataField field = schema.fields().get(i);
+ if (field.name().equals(partitionField)) {
+ rowValues.add(BinaryString.fromString(partitionValue));
+ } else if (field.type() instanceof IntType) {
+ rowValues.add(random.nextInt());
+ } else if (field.type() instanceof DoubleType) {
+ rowValues.add(random.nextDouble());
+ } else if (field.type() instanceof VarCharType) {
+ rowValues.add(BinaryString.fromString(field.name() + "_" + rowIdx + "_" + i));
+ } else if (field.type() instanceof LocalZonedTimestampType) {
+ rowValues.add(Timestamp.fromEpochMillis(System.currentTimeMillis()));
+ } else if (field.type() instanceof BooleanType) {
+ rowValues.add(random.nextBoolean());
+ } else {
+ throw new UnsupportedOperationException("Unsupported field type: " + field.type());
+ }
+ }
+
+ return GenericRow.of(rowValues.toArray());
+ }
+
+ private static String initBasePath(Path tempDir, String tableName) {
+ try {
+ Path basePath = tempDir.resolve(tableName);
+ Files.createDirectories(basePath);
+ return basePath.toUri().toString();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public List<GenericRow> insertRows(int numRows) {
+ String partitionValue = LEVEL_VALUES.get(0);
+ return insertRecordsToPartition(numRows, partitionValue);
+ }
+
+ @Override
+ public List<GenericRow> insertRecordsForSpecialPartition(int numRows) {
+ return insertRecordsToPartition(numRows, SPECIAL_PARTITION_VALUE);
+ }
+
+ private List<GenericRow> insertRecordsToPartition(int numRows, String partitionValue) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ List<GenericRow> rows = new ArrayList<>(numRows);
+ for (int i = 0; i < numRows; i++) {
+ GenericRow row = buildGenericRow(i, paimonTable.schema(), partitionValue);
+ writer.write(row);
+ rows.add(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ return rows;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to insert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void upsertRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to upsert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ row.setRowKind(RowKind.DELETE);
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete rows from Paimon table", e);
+ }
+ }
+
+ private void compactTable() {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ SnapshotReader snapshotReader = paimonTable.newSnapshotReader();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (BucketEntry bucketEntry : snapshotReader.bucketEntries()) {
+ writer.compact(bucketEntry.partition(), bucketEntry.bucket(), true);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to compact writes in Paimon table", e);
+ }
+ }
+
+ private static void commitWrites(BatchWriteBuilder batchWriteBuilder, BatchTableWrite writer)
+ throws Exception {
+ BatchTableCommit commit = batchWriteBuilder.newCommit();
+ List<CommitMessage> messages = writer.prepareCommit();
+ try {
+ commit.commit(messages);
+ } catch (Exception e) {
+ commit.abort(messages);
+ throw new RuntimeException("Failed to commit writes to Paimon table", e);
+ } finally {
+ commit.close();
+ }
+ }
+
+ @Override
+ public void deletePartition(String partitionValue) {
+ try (BatchTableCommit commit = paimonTable.newBatchWriteBuilder().newCommit()) {
+ commit.truncatePartitions(
+ ParameterUtils.getPartitions(partitionField + "=" + partitionValue));
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete partition from Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteSpecialPartition() {
+ deletePartition(SPECIAL_PARTITION_VALUE);
+ }
+
+ @Override
+ public String getBasePath() {
+ return paimonTable.location().toString();
+ }
+
+ @Override
+ public String getMetadataPath() {
+ return paimonTable.snapshotManager().snapshotDirectory().toString();
+ }
+
+ @Override
+ public String getOrderByColumn() {
+ return "id";
+ }
+
+ @Override
+ public void close() {}
+
+ @Override
+ public void reload() {}
+
+ @Override
+ public List<String> getColumnsToSelect() {
+ return paimonTable.schema().fieldNames().stream()
+ .filter(
+ // TODO Hudi thinks that paimon buckets are partition values, not sure how to handle it
+ // filtering out the partition field on the comparison for now
+ field -> !field.equals(partitionField))
Review Comment:
Yes, I think I can fix it in Hudi but it will need to wait for a future Hudi release.
Let's focus on getting Paimon to Delta & Iceberg for now?
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2410847074
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
No, I am asking about the IDs set at the parquet level
--
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] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2493393531
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonPartitionExtractor.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.paimon;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
+
+import org.apache.xtable.model.schema.InternalField;
+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;
+
+/** Extracts partition spec for Paimon as identity transforms on partition keys. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonPartitionExtractor {
+
+ private final PaimonSchemaExtractor paimonSchemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private static final PaimonPartitionExtractor INSTANCE = new PaimonPartitionExtractor();
+
+ public static PaimonPartitionExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalPartitionField> toInternalPartitionFields(
+ List<String> partitionKeys, InternalSchema schema) {
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return partitionKeys.stream()
+ .map(key -> toPartitionField(key, schema))
+ .collect(Collectors.toList());
+ }
+
+ public List<PartitionValue> toPartitionValues(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ InternalSchema internalSchema = paimonSchemaExtractor.toInternalSchema(table.schema());
+
+ List<PartitionValue> partitionValues = new ArrayList<>();
+ for (Map.Entry<String, String> entry :
+ partitionComputer.generatePartValues(partition).entrySet()) {
+ PartitionValue partitionValue =
+ PartitionValue.builder()
+ .partitionField(toPartitionField(entry.getKey(), internalSchema))
+ .range(Range.scalar(entry.getValue()))
+ .build();
+ partitionValues.add(partitionValue);
+ }
+ return partitionValues;
+ }
+
+ public Optional<String> toPartitionPath(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ return partitionComputer.generatePartValues(partition).entrySet().stream()
+ .map(e -> e.getKey() + "=" + e.getValue())
+ .reduce((a, b) -> a + "/" + b);
+ }
+
+ private InternalPartitionField toPartitionField(String key, InternalSchema schema) {
+ InternalField sourceField =
+ findField(schema, key)
+ .orElseThrow(
+ () -> new IllegalArgumentException("Partition key not found in schema: " + key));
+ return InternalPartitionField.builder()
+ .sourceField(sourceField)
+ .transformType(PartitionTransformType.VALUE)
Review Comment:
tracking here https://github.com/apache/incubator-xtable/issues/758
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2347683864
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
I've tried to offset the `fieldId` by one and I'm still getting the same class of problems, but on different fields... I'm unsure what to try next... 😞
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2334987885
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
I believe that Paimon `DataField.id` should be the field index, and I'm passing it to `InternalField.fieldId`
Where can we see the field index in the parquet file? `parquet-tools meta` or `parquet-tools schema` doesn't have any reference to index, from what I can tell 🤔
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2335002703
##########
xtable-core/pom.xml:
##########
@@ -206,6 +218,14 @@
<skip>false</skip>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>11</source>
+ <target>11</target>
Review Comment:
Okay, I'll remove this.
I wonder if it can cause problems if developers end up using features from 11 and our target is 8 🤔
--
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] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2488421998
##########
xtable-core/src/test/java/org/apache/xtable/paimon/TestPaimonSchemaExtractor.java:
##########
@@ -0,0 +1,538 @@
+/*
+ * 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.paimon;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.CharType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DateType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.FloatType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.SmallIntType;
+import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.types.TinyIntType;
+import org.apache.paimon.types.VarCharType;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+
+public class TestPaimonSchemaExtractor {
+ private static final PaimonSchemaExtractor schemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private void assertField(DataField paimonField, InternalField expectedInternalField) {
+ assertField(paimonField, expectedInternalField, Collections.emptyList());
+ }
+
+ private void assertField(
+ DataField paimonField, InternalField expectedInternalField, List<String> primaryKeys) {
+ TableSchema paimonSchema =
+ new TableSchema(
+ 0,
+ Collections.singletonList(paimonField),
+ 0,
+ Collections.emptyList(),
+ primaryKeys,
+ new HashMap<>(),
+ "");
+ InternalSchema internalSchema = schemaExtractor.toInternalSchema(paimonSchema);
+ List<InternalField> recordKeyFields =
+ primaryKeys.isEmpty()
+ ? Collections.emptyList()
+ : Collections.singletonList(expectedInternalField);
+ InternalSchema expectedSchema =
+ InternalSchema.builder()
+ .name("record")
+ .dataType(InternalType.RECORD)
+ .fields(Collections.singletonList(expectedInternalField))
+ .recordKeyFields(recordKeyFields)
+ .build();
+ assertEquals(expectedSchema, internalSchema);
+ }
+
+ @Test
+ void testCharField() {
+ DataField paimonField = new DataField(0, "char_field", new CharType(10));
+ InternalField expectedField =
+ InternalField.builder()
+ .name("char_field")
+ .fieldId(0)
+ .schema(
+ InternalSchema.builder()
+ .name("CHAR(10)")
+ .dataType(InternalType.STRING)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testVarcharField() {
+ DataField paimonField = new DataField(1, "varchar_field", new VarCharType(255));
+ InternalField expectedField =
+ InternalField.builder()
+ .name("varchar_field")
+ .fieldId(1)
+ .schema(
+ InternalSchema.builder()
+ .name("VARCHAR(255)")
+ .dataType(InternalType.STRING)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testBooleanField() {
+ DataField paimonField = new DataField(2, "boolean_field", new BooleanType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("boolean_field")
+ .fieldId(2)
+ .schema(
+ InternalSchema.builder()
+ .name("BOOLEAN")
+ .dataType(InternalType.BOOLEAN)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testTinyIntField() {
+ DataField paimonField = new DataField(3, "tinyint_field", new TinyIntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("tinyint_field")
+ .fieldId(3)
+ .schema(
+ InternalSchema.builder()
+ .name("TINYINT")
+ .dataType(InternalType.INT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testSmallIntField() {
+ DataField paimonField = new DataField(4, "smallint_field", new SmallIntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("smallint_field")
+ .fieldId(4)
+ .schema(
+ InternalSchema.builder()
+ .name("SMALLINT")
+ .dataType(InternalType.INT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testIntField() {
+ DataField paimonField = new DataField(5, "int_field", new IntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("int_field")
+ .fieldId(5)
+ .schema(
+ InternalSchema.builder()
+ .name("INT")
+ .dataType(InternalType.INT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testBigIntField() {
+ DataField paimonField = new DataField(6, "bigint_field", new BigIntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("bigint_field")
+ .fieldId(6)
+ .schema(
+ InternalSchema.builder()
+ .name("BIGINT")
+ .dataType(InternalType.LONG)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testFloatField() {
+ DataField paimonField = new DataField(7, "float_field", new FloatType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("float_field")
+ .fieldId(7)
+ .schema(
+ InternalSchema.builder()
+ .name("FLOAT")
+ .dataType(InternalType.FLOAT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testDoubleField() {
+ DataField paimonField = new DataField(8, "double_field", new DoubleType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("double_field")
+ .fieldId(8)
+ .schema(
+ InternalSchema.builder()
+ .name("DOUBLE")
+ .dataType(InternalType.DOUBLE)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testDateField() {
+ DataField paimonField = new DataField(9, "date_field", new DateType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("date_field")
+ .fieldId(9)
+ .schema(
+ InternalSchema.builder()
+ .name("DATE")
+ .dataType(InternalType.DATE)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9})
+ void testTimestampField(int precision) {
+ DataField paimonField = new DataField(10, "timestamp_field", new TimestampType(precision));
+ Map<InternalSchema.MetadataKey, Object> timestampMetadata =
+ Collections.singletonMap(
+ InternalSchema.MetadataKey.TIMESTAMP_PRECISION, InternalSchema.MetadataValue.MICROS);
Review Comment:
good catch, fixed!
--
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] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2484419428
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonDataFileExtractor.java:
##########
@@ -0,0 +1,94 @@
+/*
+ * 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.paimon;
+
+import java.util.*;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+
+import org.apache.xtable.model.stat.ColumnStat;
+import org.apache.xtable.model.storage.InternalDataFile;
+
+public class PaimonDataFileExtractor {
+
+ private final PaimonPartitionExtractor partitionExtractor =
+ PaimonPartitionExtractor.getInstance();
+
+ private static final PaimonDataFileExtractor INSTANCE = new PaimonDataFileExtractor();
+
+ public static PaimonDataFileExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalDataFile> toInternalDataFiles(FileStoreTable table, Snapshot snapshot) {
+ List<InternalDataFile> result = new ArrayList<>();
+ Iterator<ManifestEntry> manifestEntryIterator =
+ newSnapshotReader(table, snapshot).readFileIterator();
+ while (manifestEntryIterator.hasNext()) {
+ result.add(toInternalDataFile(table, manifestEntryIterator.next()));
+ }
+ return result;
+ }
+
+ private InternalDataFile toInternalDataFile(FileStoreTable table, ManifestEntry entry) {
+ return InternalDataFile.builder()
+ .physicalPath(toFullPhysicalPath(table, entry))
+ .fileSizeBytes(entry.file().fileSize())
+ .lastModified(entry.file().creationTimeEpochMillis())
+ .recordCount(entry.file().rowCount())
+ .partitionValues(partitionExtractor.toPartitionValues(table, entry.partition()))
+ .columnStats(toColumnStats(entry.file()))
+ .build();
+ }
+
+ private String toFullPhysicalPath(FileStoreTable table, ManifestEntry entry) {
+ String basePath = table.location().toString();
+ String bucketPath = "bucket-" + entry.bucket();
+ String filePath = entry.file().fileName();
+
+ Optional<String> partitionPath = partitionExtractor.toPartitionPath(table, entry.partition());
+ if (partitionPath.isPresent()) {
+ return String.join("/", basePath, partitionPath.get(), bucketPath, filePath);
+ } else {
+ return String.join("/", basePath, bucketPath, filePath);
+ }
+ }
+
+ private List<ColumnStat> toColumnStats(DataFileMeta file) {
+ // TODO: Implement logic to extract column stats from the file meta
+ return Collections.emptyList();
+ }
+
+ private SnapshotReader newSnapshotReader(FileStoreTable table, Snapshot snapshot) {
+ // If the table has primary keys, we read only the top level files
Review Comment:
Yes, they are similar: https://paimon.apache.org/docs/1.2/primary-key-table/overview/#sorted-runs
--
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] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2484841927
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonPartitionExtractor.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.paimon;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
+
+import org.apache.xtable.model.schema.InternalField;
+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;
+
+/** Extracts partition spec for Paimon as identity transforms on partition keys. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonPartitionExtractor {
+
+ private final PaimonSchemaExtractor paimonSchemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private static final PaimonPartitionExtractor INSTANCE = new PaimonPartitionExtractor();
+
+ public static PaimonPartitionExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalPartitionField> toInternalPartitionFields(
+ List<String> partitionKeys, InternalSchema schema) {
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return partitionKeys.stream()
+ .map(key -> toPartitionField(key, schema))
+ .collect(Collectors.toList());
+ }
+
+ public List<PartitionValue> toPartitionValues(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ InternalSchema internalSchema = paimonSchemaExtractor.toInternalSchema(table.schema());
+
+ List<PartitionValue> partitionValues = new ArrayList<>();
+ for (Map.Entry<String, String> entry :
+ partitionComputer.generatePartValues(partition).entrySet()) {
+ PartitionValue partitionValue =
+ PartitionValue.builder()
+ .partitionField(toPartitionField(entry.getKey(), internalSchema))
+ .range(Range.scalar(entry.getValue()))
+ .build();
+ partitionValues.add(partitionValue);
+ }
+ return partitionValues;
+ }
+
+ public Optional<String> toPartitionPath(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ return partitionComputer.generatePartValues(partition).entrySet().stream()
+ .map(e -> e.getKey() + "=" + e.getValue())
+ .reduce((a, b) -> a + "/" + b);
+ }
+
+ private InternalPartitionField toPartitionField(String key, InternalSchema schema) {
+ InternalField sourceField =
+ findField(schema, key)
+ .orElseThrow(
+ () -> new IllegalArgumentException("Partition key not found in schema: " + key));
+ return InternalPartitionField.builder()
+ .sourceField(sourceField)
+ .transformType(PartitionTransformType.VALUE)
Review Comment:
Ok, we'll want to add support for this as well. Do you want to do it in this PR or as part of a follow up task?
--
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] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2485076603
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonPartitionExtractor.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.paimon;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
+
+import org.apache.xtable.model.schema.InternalField;
+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;
+
+/** Extracts partition spec for Paimon as identity transforms on partition keys. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonPartitionExtractor {
+
+ private final PaimonSchemaExtractor paimonSchemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private static final PaimonPartitionExtractor INSTANCE = new PaimonPartitionExtractor();
+
+ public static PaimonPartitionExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalPartitionField> toInternalPartitionFields(
+ List<String> partitionKeys, InternalSchema schema) {
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return partitionKeys.stream()
+ .map(key -> toPartitionField(key, schema))
+ .collect(Collectors.toList());
+ }
+
+ public List<PartitionValue> toPartitionValues(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ InternalSchema internalSchema = paimonSchemaExtractor.toInternalSchema(table.schema());
+
+ List<PartitionValue> partitionValues = new ArrayList<>();
+ for (Map.Entry<String, String> entry :
+ partitionComputer.generatePartValues(partition).entrySet()) {
+ PartitionValue partitionValue =
+ PartitionValue.builder()
+ .partitionField(toPartitionField(entry.getKey(), internalSchema))
+ .range(Range.scalar(entry.getValue()))
+ .build();
+ partitionValues.add(partitionValue);
+ }
+ return partitionValues;
+ }
+
+ public Optional<String> toPartitionPath(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ return partitionComputer.generatePartValues(partition).entrySet().stream()
+ .map(e -> e.getKey() + "=" + e.getValue())
+ .reduce((a, b) -> a + "/" + b);
+ }
+
+ private InternalPartitionField toPartitionField(String key, InternalSchema schema) {
+ InternalField sourceField =
+ findField(schema, key)
+ .orElseThrow(
+ () -> new IllegalArgumentException("Partition key not found in schema: " + key));
+ return InternalPartitionField.builder()
+ .sourceField(sourceField)
+ .transformType(PartitionTransformType.VALUE)
Review Comment:
I'm not sure if I understand why partitions need to have a type in XTable... In my mind, XTable could simply treat all partition values equally as opaque values without having to understand if they are dates or timestamps, or anything really. Is there a document where I can read to better understand the rationale behind partition types?
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2404758916
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
Where do the field IDs get set in Paimon?
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2331953293
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
Thanks @the-other-tim-brown, that was a good insight!
I've added the special fields based on `org.apache.paimon.table.SpecialFields` helper and now the InternalSchema matches the parquet files.
<img width="521" height="293" alt="image" src="https://github.com/user-attachments/assets/d5938820-1268-4cd3-a12a-d2d49cebb5e8" />
<img width="1291" height="834" alt="image" src="https://github.com/user-attachments/assets/64ec0f4c-7fab-4d66-9d39-ddaab3ad344d" />
with that, the error changed to
```
class org.apache.iceberg.shaded.org.apache.arrow.vector.VarCharVector cannot be cast to class org.apache.iceberg.shaded.org.apache.arrow.vector.BitVector
```
This indicates that it's still the same class of problem, but it also indicates that the order of fields here matters for the iceberg reader.
I've tried sorting by `name` and `fieldId` but I'm still getting variations of the same problem... :(
Any more ideas? 😅
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2331954728
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
+ private static final PaimonSchemaExtractor INSTANCE = new PaimonSchemaExtractor();
+
+ public static PaimonSchemaExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public InternalSchema toInternalSchema(TableSchema paimonSchema) {
+ RowType rowType = paimonSchema.logicalRowType();
+ List<InternalField> fields = toInternalFields(rowType);
+ return InternalSchema.builder()
+ .name("record")
+ .dataType(InternalType.RECORD)
+ .fields(fields)
+ .recordKeyFields(primaryKeyFields(paimonSchema, fields))
+ .build();
+ }
+
+ private List<InternalField> primaryKeyFields(
+ TableSchema paimonSchema, List<InternalField> internalFields) {
+ List<String> keys = paimonSchema.primaryKeys();
Review Comment:
that is a good question, I'll double check
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2331973890
##########
xtable-core/pom.xml:
##########
@@ -206,6 +218,14 @@
<skip>false</skip>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>11</source>
+ <target>11</target>
Review Comment:
I'm getting these compatibility errors when using java 8 on my machine, so I thought xtable was on 11 already but wasn't configured
<img width="1105" height="161" alt="image" src="https://github.com/user-attachments/assets/0c9d224a-a729-4381-ae40-4e89eb8e265b" />
maybe that's an issue with my intellij, as it runs fine with
<img width="1639" height="96" alt="image" src="https://github.com/user-attachments/assets/3d25d9f3-b1fe-4548-806b-756af482bf71" />
everything works fine on 11 🤔
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2335064487
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
`parquet-tools schema` gives me the field ID when I run I run 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] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2491232152
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonPartitionExtractor.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.paimon;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
+
+import org.apache.xtable.model.schema.InternalField;
+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;
+
+/** Extracts partition spec for Paimon as identity transforms on partition keys. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonPartitionExtractor {
+
+ private final PaimonSchemaExtractor paimonSchemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private static final PaimonPartitionExtractor INSTANCE = new PaimonPartitionExtractor();
+
+ public static PaimonPartitionExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalPartitionField> toInternalPartitionFields(
+ List<String> partitionKeys, InternalSchema schema) {
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return partitionKeys.stream()
+ .map(key -> toPartitionField(key, schema))
+ .collect(Collectors.toList());
+ }
+
+ public List<PartitionValue> toPartitionValues(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ InternalSchema internalSchema = paimonSchemaExtractor.toInternalSchema(table.schema());
+
+ List<PartitionValue> partitionValues = new ArrayList<>();
+ for (Map.Entry<String, String> entry :
+ partitionComputer.generatePartValues(partition).entrySet()) {
+ PartitionValue partitionValue =
+ PartitionValue.builder()
+ .partitionField(toPartitionField(entry.getKey(), internalSchema))
+ .range(Range.scalar(entry.getValue()))
+ .build();
+ partitionValues.add(partitionValue);
+ }
+ return partitionValues;
+ }
+
+ public Optional<String> toPartitionPath(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ return partitionComputer.generatePartValues(partition).entrySet().stream()
+ .map(e -> e.getKey() + "=" + e.getValue())
+ .reduce((a, b) -> a + "/" + b);
+ }
+
+ private InternalPartitionField toPartitionField(String key, InternalSchema schema) {
+ InternalField sourceField =
+ findField(schema, key)
+ .orElseThrow(
+ () -> new IllegalArgumentException("Partition key not found in schema: " + key));
+ return InternalPartitionField.builder()
+ .sourceField(sourceField)
+ .transformType(PartitionTransformType.VALUE)
Review Comment:
Yes, it's fine to tackle it as a follow up. Let's just make sure there is a GH issue for it so we can potentially spread out the work to other contributors as well.
Hudi and Delta have similar concepts but are not as well documented. You must specify some more context to ensure that the partition pruning kicks in.
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2425006253
##########
xtable-core/src/test/java/org/apache/xtable/TestPaimonTable.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.manifest.BucketEntry;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.ParameterUtils;
+
+public class TestPaimonTable implements GenericTable<GenericRow, String> {
+
+ private final Random random = new Random();
+ private final FileStoreTable paimonTable;
+ private final String partitionField;
+
+ public TestPaimonTable(FileStoreTable paimonTable, String partitionField) {
+ this.paimonTable = paimonTable;
+ this.partitionField = partitionField;
+ }
+
+ public static GenericTable<GenericRow, String> createTable(
+ String tableName,
+ String partitionField,
+ Path tempDir,
+ Configuration hadoopConf,
+ boolean additionalColumns) {
+ String basePath = initBasePath(tempDir, tableName);
+ Catalog catalog = createFilesystemCatalog(basePath, hadoopConf);
+ FileStoreTable paimonTable = createTable(catalog, partitionField, additionalColumns);
+
+ System.out.println(
+ "Initialized Paimon test table at base path: "
+ + basePath
+ + " with partition field: "
+ + partitionField
+ + " and additional columns: "
+ + additionalColumns);
+
+ return new TestPaimonTable(paimonTable, partitionField);
+ }
+
+ public static Catalog createFilesystemCatalog(String basePath, Configuration hadoopConf) {
+ CatalogContext context = CatalogContext.create(new org.apache.paimon.fs.Path(basePath));
+ return CatalogFactory.createCatalog(context);
+ }
+
+ public static FileStoreTable createTable(
+ Catalog catalog, String partitionField, boolean additionalColumns) {
+ try {
+ catalog.createDatabase("test_db", true);
+ Identifier identifier = Identifier.create("test_db", "test_table");
+ Schema schema = buildSchema(partitionField, additionalColumns);
+ catalog.createTable(identifier, schema, true);
+ return (FileStoreTable) catalog.getTable(identifier);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static Schema buildSchema(String partitionField, boolean additionalColumns) {
+ Schema.Builder builder =
+ Schema.newBuilder()
+ .primaryKey("id")
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column("value", DataTypes.DOUBLE())
+ .column("created_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("updated_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("is_active", DataTypes.BOOLEAN())
+ .column("description", DataTypes.VARCHAR(255))
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .option("full-compaction.delta-commits", "1");
+
+ if (partitionField != null) {
+ builder
+ .primaryKey("id", partitionField)
+ .column(partitionField, DataTypes.STRING())
+ .partitionKeys(partitionField);
+ }
+
+ if (additionalColumns) {
+ builder.column("extra_info", DataTypes.STRING()).column("extra_value", DataTypes.DOUBLE());
+ }
+
+ return builder.build();
+ }
+
+ private GenericRow buildGenericRow(int rowIdx, TableSchema schema, String partitionValue) {
+ List<Object> rowValues = new ArrayList<>(schema.fields().size());
+ for (int i = 0; i < schema.fields().size(); i++) {
+ DataField field = schema.fields().get(i);
+ if (field.name().equals(partitionField)) {
+ rowValues.add(BinaryString.fromString(partitionValue));
+ } else if (field.type() instanceof IntType) {
+ rowValues.add(random.nextInt());
+ } else if (field.type() instanceof DoubleType) {
+ rowValues.add(random.nextDouble());
+ } else if (field.type() instanceof VarCharType) {
+ rowValues.add(BinaryString.fromString(field.name() + "_" + rowIdx + "_" + i));
+ } else if (field.type() instanceof LocalZonedTimestampType) {
+ rowValues.add(Timestamp.fromEpochMillis(System.currentTimeMillis()));
+ } else if (field.type() instanceof BooleanType) {
+ rowValues.add(random.nextBoolean());
+ } else {
+ throw new UnsupportedOperationException("Unsupported field type: " + field.type());
+ }
+ }
+
+ return GenericRow.of(rowValues.toArray());
+ }
+
+ private static String initBasePath(Path tempDir, String tableName) {
+ try {
+ Path basePath = tempDir.resolve(tableName);
+ Files.createDirectories(basePath);
+ return basePath.toUri().toString();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public List<GenericRow> insertRows(int numRows) {
+ String partitionValue = LEVEL_VALUES.get(0);
+ return insertRecordsToPartition(numRows, partitionValue);
+ }
+
+ @Override
+ public List<GenericRow> insertRecordsForSpecialPartition(int numRows) {
+ return insertRecordsToPartition(numRows, SPECIAL_PARTITION_VALUE);
+ }
+
+ private List<GenericRow> insertRecordsToPartition(int numRows, String partitionValue) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ List<GenericRow> rows = new ArrayList<>(numRows);
+ for (int i = 0; i < numRows; i++) {
+ GenericRow row = buildGenericRow(i, paimonTable.schema(), partitionValue);
+ writer.write(row);
+ rows.add(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ return rows;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to insert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void upsertRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to upsert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ row.setRowKind(RowKind.DELETE);
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete rows from Paimon table", e);
+ }
+ }
+
+ private void compactTable() {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ SnapshotReader snapshotReader = paimonTable.newSnapshotReader();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (BucketEntry bucketEntry : snapshotReader.bucketEntries()) {
+ writer.compact(bucketEntry.partition(), bucketEntry.bucket(), true);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to compact writes in Paimon table", e);
+ }
+ }
+
+ private static void commitWrites(BatchWriteBuilder batchWriteBuilder, BatchTableWrite writer)
+ throws Exception {
+ BatchTableCommit commit = batchWriteBuilder.newCommit();
+ List<CommitMessage> messages = writer.prepareCommit();
+ try {
+ commit.commit(messages);
+ } catch (Exception e) {
+ commit.abort(messages);
+ throw new RuntimeException("Failed to commit writes to Paimon table", e);
+ } finally {
+ commit.close();
+ }
+ }
+
+ @Override
+ public void deletePartition(String partitionValue) {
+ try (BatchTableCommit commit = paimonTable.newBatchWriteBuilder().newCommit()) {
+ commit.truncatePartitions(
+ ParameterUtils.getPartitions(partitionField + "=" + partitionValue));
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete partition from Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteSpecialPartition() {
+ deletePartition(SPECIAL_PARTITION_VALUE);
+ }
+
+ @Override
+ public String getBasePath() {
+ return paimonTable.location().toString();
+ }
+
+ @Override
+ public String getMetadataPath() {
+ return paimonTable.snapshotManager().snapshotDirectory().toString();
+ }
+
+ @Override
+ public String getOrderByColumn() {
+ return "id";
+ }
+
+ @Override
+ public void close() {}
+
+ @Override
+ public void reload() {}
+
+ @Override
+ public List<String> getColumnsToSelect() {
+ return paimonTable.schema().fieldNames().stream()
+ .filter(
+ // TODO Hudi thinks that paimon buckets are partition values, not sure how to handle it
+ // filtering out the partition field on the comparison for now
+ field -> !field.equals(partitionField))
Review Comment:
@the-other-tim-brown do you have any opinions about this?
> Paimon has buckets as another folder division within partitions (e.g. `partition=2025-09-01/bucket-0/data-file.parquet`), however, Hudi considers the bucket value as part of the partition value (e.g. `2025-09-01/bucket-0`). Looking at the code, it seems that directory structure == partition values assumption is pretty deep, so I wonder if there is a way to work around that, or we should call out as a limitation between both formats.
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2429590531
##########
xtable-core/src/test/java/org/apache/xtable/TestPaimonTable.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.manifest.BucketEntry;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.ParameterUtils;
+
+public class TestPaimonTable implements GenericTable<GenericRow, String> {
+
+ private final Random random = new Random();
+ private final FileStoreTable paimonTable;
+ private final String partitionField;
+
+ public TestPaimonTable(FileStoreTable paimonTable, String partitionField) {
+ this.paimonTable = paimonTable;
+ this.partitionField = partitionField;
+ }
+
+ public static GenericTable<GenericRow, String> createTable(
+ String tableName,
+ String partitionField,
+ Path tempDir,
+ Configuration hadoopConf,
+ boolean additionalColumns) {
+ String basePath = initBasePath(tempDir, tableName);
+ Catalog catalog = createFilesystemCatalog(basePath, hadoopConf);
+ FileStoreTable paimonTable = createTable(catalog, partitionField, additionalColumns);
+
+ System.out.println(
+ "Initialized Paimon test table at base path: "
+ + basePath
+ + " with partition field: "
+ + partitionField
+ + " and additional columns: "
+ + additionalColumns);
+
+ return new TestPaimonTable(paimonTable, partitionField);
+ }
+
+ public static Catalog createFilesystemCatalog(String basePath, Configuration hadoopConf) {
+ CatalogContext context = CatalogContext.create(new org.apache.paimon.fs.Path(basePath));
+ return CatalogFactory.createCatalog(context);
+ }
+
+ public static FileStoreTable createTable(
+ Catalog catalog, String partitionField, boolean additionalColumns) {
+ try {
+ catalog.createDatabase("test_db", true);
+ Identifier identifier = Identifier.create("test_db", "test_table");
+ Schema schema = buildSchema(partitionField, additionalColumns);
+ catalog.createTable(identifier, schema, true);
+ return (FileStoreTable) catalog.getTable(identifier);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static Schema buildSchema(String partitionField, boolean additionalColumns) {
+ Schema.Builder builder =
+ Schema.newBuilder()
+ .primaryKey("id")
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column("value", DataTypes.DOUBLE())
+ .column("created_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("updated_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("is_active", DataTypes.BOOLEAN())
+ .column("description", DataTypes.VARCHAR(255))
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .option("full-compaction.delta-commits", "1");
+
+ if (partitionField != null) {
+ builder
+ .primaryKey("id", partitionField)
+ .column(partitionField, DataTypes.STRING())
+ .partitionKeys(partitionField);
+ }
+
+ if (additionalColumns) {
+ builder.column("extra_info", DataTypes.STRING()).column("extra_value", DataTypes.DOUBLE());
+ }
+
+ return builder.build();
+ }
+
+ private GenericRow buildGenericRow(int rowIdx, TableSchema schema, String partitionValue) {
+ List<Object> rowValues = new ArrayList<>(schema.fields().size());
+ for (int i = 0; i < schema.fields().size(); i++) {
+ DataField field = schema.fields().get(i);
+ if (field.name().equals(partitionField)) {
+ rowValues.add(BinaryString.fromString(partitionValue));
+ } else if (field.type() instanceof IntType) {
+ rowValues.add(random.nextInt());
+ } else if (field.type() instanceof DoubleType) {
+ rowValues.add(random.nextDouble());
+ } else if (field.type() instanceof VarCharType) {
+ rowValues.add(BinaryString.fromString(field.name() + "_" + rowIdx + "_" + i));
+ } else if (field.type() instanceof LocalZonedTimestampType) {
+ rowValues.add(Timestamp.fromEpochMillis(System.currentTimeMillis()));
+ } else if (field.type() instanceof BooleanType) {
+ rowValues.add(random.nextBoolean());
+ } else {
+ throw new UnsupportedOperationException("Unsupported field type: " + field.type());
+ }
+ }
+
+ return GenericRow.of(rowValues.toArray());
+ }
+
+ private static String initBasePath(Path tempDir, String tableName) {
+ try {
+ Path basePath = tempDir.resolve(tableName);
+ Files.createDirectories(basePath);
+ return basePath.toUri().toString();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public List<GenericRow> insertRows(int numRows) {
+ String partitionValue = LEVEL_VALUES.get(0);
+ return insertRecordsToPartition(numRows, partitionValue);
+ }
+
+ @Override
+ public List<GenericRow> insertRecordsForSpecialPartition(int numRows) {
+ return insertRecordsToPartition(numRows, SPECIAL_PARTITION_VALUE);
+ }
+
+ private List<GenericRow> insertRecordsToPartition(int numRows, String partitionValue) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ List<GenericRow> rows = new ArrayList<>(numRows);
+ for (int i = 0; i < numRows; i++) {
+ GenericRow row = buildGenericRow(i, paimonTable.schema(), partitionValue);
+ writer.write(row);
+ rows.add(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ return rows;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to insert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void upsertRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to upsert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ row.setRowKind(RowKind.DELETE);
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete rows from Paimon table", e);
+ }
+ }
+
+ private void compactTable() {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ SnapshotReader snapshotReader = paimonTable.newSnapshotReader();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (BucketEntry bucketEntry : snapshotReader.bucketEntries()) {
+ writer.compact(bucketEntry.partition(), bucketEntry.bucket(), true);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to compact writes in Paimon table", e);
+ }
+ }
+
+ private static void commitWrites(BatchWriteBuilder batchWriteBuilder, BatchTableWrite writer)
+ throws Exception {
+ BatchTableCommit commit = batchWriteBuilder.newCommit();
+ List<CommitMessage> messages = writer.prepareCommit();
+ try {
+ commit.commit(messages);
+ } catch (Exception e) {
+ commit.abort(messages);
+ throw new RuntimeException("Failed to commit writes to Paimon table", e);
+ } finally {
+ commit.close();
+ }
+ }
+
+ @Override
+ public void deletePartition(String partitionValue) {
+ try (BatchTableCommit commit = paimonTable.newBatchWriteBuilder().newCommit()) {
+ commit.truncatePartitions(
+ ParameterUtils.getPartitions(partitionField + "=" + partitionValue));
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete partition from Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteSpecialPartition() {
+ deletePartition(SPECIAL_PARTITION_VALUE);
+ }
+
+ @Override
+ public String getBasePath() {
+ return paimonTable.location().toString();
+ }
+
+ @Override
+ public String getMetadataPath() {
+ return paimonTable.snapshotManager().snapshotDirectory().toString();
+ }
+
+ @Override
+ public String getOrderByColumn() {
+ return "id";
+ }
+
+ @Override
+ public void close() {}
+
+ @Override
+ public void reload() {}
+
+ @Override
+ public List<String> getColumnsToSelect() {
+ return paimonTable.schema().fieldNames().stream()
+ .filter(
+ // TODO Hudi thinks that paimon buckets are partition values, not sure how to handle it
+ // filtering out the partition field on the comparison for now
+ field -> !field.equals(partitionField))
Review Comment:
Is this an issue in the Apache Hudi or XTable repo?
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2430870597
##########
xtable-core/src/test/java/org/apache/xtable/TestPaimonTable.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.manifest.BucketEntry;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.ParameterUtils;
+
+public class TestPaimonTable implements GenericTable<GenericRow, String> {
+
+ private final Random random = new Random();
+ private final FileStoreTable paimonTable;
+ private final String partitionField;
+
+ public TestPaimonTable(FileStoreTable paimonTable, String partitionField) {
+ this.paimonTable = paimonTable;
+ this.partitionField = partitionField;
+ }
+
+ public static GenericTable<GenericRow, String> createTable(
+ String tableName,
+ String partitionField,
+ Path tempDir,
+ Configuration hadoopConf,
+ boolean additionalColumns) {
+ String basePath = initBasePath(tempDir, tableName);
+ Catalog catalog = createFilesystemCatalog(basePath, hadoopConf);
+ FileStoreTable paimonTable = createTable(catalog, partitionField, additionalColumns);
+
+ System.out.println(
+ "Initialized Paimon test table at base path: "
+ + basePath
+ + " with partition field: "
+ + partitionField
+ + " and additional columns: "
+ + additionalColumns);
+
+ return new TestPaimonTable(paimonTable, partitionField);
+ }
+
+ public static Catalog createFilesystemCatalog(String basePath, Configuration hadoopConf) {
+ CatalogContext context = CatalogContext.create(new org.apache.paimon.fs.Path(basePath));
+ return CatalogFactory.createCatalog(context);
+ }
+
+ public static FileStoreTable createTable(
+ Catalog catalog, String partitionField, boolean additionalColumns) {
+ try {
+ catalog.createDatabase("test_db", true);
+ Identifier identifier = Identifier.create("test_db", "test_table");
+ Schema schema = buildSchema(partitionField, additionalColumns);
+ catalog.createTable(identifier, schema, true);
+ return (FileStoreTable) catalog.getTable(identifier);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static Schema buildSchema(String partitionField, boolean additionalColumns) {
+ Schema.Builder builder =
+ Schema.newBuilder()
+ .primaryKey("id")
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column("value", DataTypes.DOUBLE())
+ .column("created_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("updated_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("is_active", DataTypes.BOOLEAN())
+ .column("description", DataTypes.VARCHAR(255))
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .option("full-compaction.delta-commits", "1");
+
+ if (partitionField != null) {
+ builder
+ .primaryKey("id", partitionField)
+ .column(partitionField, DataTypes.STRING())
+ .partitionKeys(partitionField);
+ }
+
+ if (additionalColumns) {
+ builder.column("extra_info", DataTypes.STRING()).column("extra_value", DataTypes.DOUBLE());
+ }
+
+ return builder.build();
+ }
+
+ private GenericRow buildGenericRow(int rowIdx, TableSchema schema, String partitionValue) {
+ List<Object> rowValues = new ArrayList<>(schema.fields().size());
+ for (int i = 0; i < schema.fields().size(); i++) {
+ DataField field = schema.fields().get(i);
+ if (field.name().equals(partitionField)) {
+ rowValues.add(BinaryString.fromString(partitionValue));
+ } else if (field.type() instanceof IntType) {
+ rowValues.add(random.nextInt());
+ } else if (field.type() instanceof DoubleType) {
+ rowValues.add(random.nextDouble());
+ } else if (field.type() instanceof VarCharType) {
+ rowValues.add(BinaryString.fromString(field.name() + "_" + rowIdx + "_" + i));
+ } else if (field.type() instanceof LocalZonedTimestampType) {
+ rowValues.add(Timestamp.fromEpochMillis(System.currentTimeMillis()));
+ } else if (field.type() instanceof BooleanType) {
+ rowValues.add(random.nextBoolean());
+ } else {
+ throw new UnsupportedOperationException("Unsupported field type: " + field.type());
+ }
+ }
+
+ return GenericRow.of(rowValues.toArray());
+ }
+
+ private static String initBasePath(Path tempDir, String tableName) {
+ try {
+ Path basePath = tempDir.resolve(tableName);
+ Files.createDirectories(basePath);
+ return basePath.toUri().toString();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public List<GenericRow> insertRows(int numRows) {
+ String partitionValue = LEVEL_VALUES.get(0);
+ return insertRecordsToPartition(numRows, partitionValue);
+ }
+
+ @Override
+ public List<GenericRow> insertRecordsForSpecialPartition(int numRows) {
+ return insertRecordsToPartition(numRows, SPECIAL_PARTITION_VALUE);
+ }
+
+ private List<GenericRow> insertRecordsToPartition(int numRows, String partitionValue) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ List<GenericRow> rows = new ArrayList<>(numRows);
+ for (int i = 0; i < numRows; i++) {
+ GenericRow row = buildGenericRow(i, paimonTable.schema(), partitionValue);
+ writer.write(row);
+ rows.add(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ return rows;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to insert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void upsertRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to upsert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ row.setRowKind(RowKind.DELETE);
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete rows from Paimon table", e);
+ }
+ }
+
+ private void compactTable() {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ SnapshotReader snapshotReader = paimonTable.newSnapshotReader();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (BucketEntry bucketEntry : snapshotReader.bucketEntries()) {
+ writer.compact(bucketEntry.partition(), bucketEntry.bucket(), true);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to compact writes in Paimon table", e);
+ }
+ }
+
+ private static void commitWrites(BatchWriteBuilder batchWriteBuilder, BatchTableWrite writer)
+ throws Exception {
+ BatchTableCommit commit = batchWriteBuilder.newCommit();
+ List<CommitMessage> messages = writer.prepareCommit();
+ try {
+ commit.commit(messages);
+ } catch (Exception e) {
+ commit.abort(messages);
+ throw new RuntimeException("Failed to commit writes to Paimon table", e);
+ } finally {
+ commit.close();
+ }
+ }
+
+ @Override
+ public void deletePartition(String partitionValue) {
+ try (BatchTableCommit commit = paimonTable.newBatchWriteBuilder().newCommit()) {
+ commit.truncatePartitions(
+ ParameterUtils.getPartitions(partitionField + "=" + partitionValue));
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete partition from Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteSpecialPartition() {
+ deletePartition(SPECIAL_PARTITION_VALUE);
+ }
+
+ @Override
+ public String getBasePath() {
+ return paimonTable.location().toString();
+ }
+
+ @Override
+ public String getMetadataPath() {
+ return paimonTable.snapshotManager().snapshotDirectory().toString();
+ }
+
+ @Override
+ public String getOrderByColumn() {
+ return "id";
+ }
+
+ @Override
+ public void close() {}
+
+ @Override
+ public void reload() {}
+
+ @Override
+ public List<String> getColumnsToSelect() {
+ return paimonTable.schema().fieldNames().stream()
+ .filter(
+ // TODO Hudi thinks that paimon buckets are partition values, not sure how to handle it
+ // filtering out the partition field on the comparison for now
+ field -> !field.equals(partitionField))
Review Comment:
As far as I can see, it starts on the XTable repo, particularly on the `HudiPathUtils.getPartitionPath`, used in many key areas of the codebase:
<img width="738" height="418" alt="image" src="https://github.com/user-attachments/assets/f1ef3272-c04d-4815-98a0-4a3c95c3c9da" />
But I'm not familiar with Hudi enough to determine if that's a limitation there as well.
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2430890371
##########
xtable-core/src/test/java/org/apache/xtable/TestPaimonTable.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.manifest.BucketEntry;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.ParameterUtils;
+
+public class TestPaimonTable implements GenericTable<GenericRow, String> {
+
+ private final Random random = new Random();
+ private final FileStoreTable paimonTable;
+ private final String partitionField;
+
+ public TestPaimonTable(FileStoreTable paimonTable, String partitionField) {
+ this.paimonTable = paimonTable;
+ this.partitionField = partitionField;
+ }
+
+ public static GenericTable<GenericRow, String> createTable(
+ String tableName,
+ String partitionField,
+ Path tempDir,
+ Configuration hadoopConf,
+ boolean additionalColumns) {
+ String basePath = initBasePath(tempDir, tableName);
+ Catalog catalog = createFilesystemCatalog(basePath, hadoopConf);
+ FileStoreTable paimonTable = createTable(catalog, partitionField, additionalColumns);
+
+ System.out.println(
+ "Initialized Paimon test table at base path: "
+ + basePath
+ + " with partition field: "
+ + partitionField
+ + " and additional columns: "
+ + additionalColumns);
+
+ return new TestPaimonTable(paimonTable, partitionField);
+ }
+
+ public static Catalog createFilesystemCatalog(String basePath, Configuration hadoopConf) {
+ CatalogContext context = CatalogContext.create(new org.apache.paimon.fs.Path(basePath));
+ return CatalogFactory.createCatalog(context);
+ }
+
+ public static FileStoreTable createTable(
+ Catalog catalog, String partitionField, boolean additionalColumns) {
+ try {
+ catalog.createDatabase("test_db", true);
+ Identifier identifier = Identifier.create("test_db", "test_table");
+ Schema schema = buildSchema(partitionField, additionalColumns);
+ catalog.createTable(identifier, schema, true);
+ return (FileStoreTable) catalog.getTable(identifier);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static Schema buildSchema(String partitionField, boolean additionalColumns) {
+ Schema.Builder builder =
+ Schema.newBuilder()
+ .primaryKey("id")
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column("value", DataTypes.DOUBLE())
+ .column("created_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("updated_at", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("is_active", DataTypes.BOOLEAN())
+ .column("description", DataTypes.VARCHAR(255))
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .option("full-compaction.delta-commits", "1");
+
+ if (partitionField != null) {
+ builder
+ .primaryKey("id", partitionField)
+ .column(partitionField, DataTypes.STRING())
+ .partitionKeys(partitionField);
+ }
+
+ if (additionalColumns) {
+ builder.column("extra_info", DataTypes.STRING()).column("extra_value", DataTypes.DOUBLE());
+ }
+
+ return builder.build();
+ }
+
+ private GenericRow buildGenericRow(int rowIdx, TableSchema schema, String partitionValue) {
+ List<Object> rowValues = new ArrayList<>(schema.fields().size());
+ for (int i = 0; i < schema.fields().size(); i++) {
+ DataField field = schema.fields().get(i);
+ if (field.name().equals(partitionField)) {
+ rowValues.add(BinaryString.fromString(partitionValue));
+ } else if (field.type() instanceof IntType) {
+ rowValues.add(random.nextInt());
+ } else if (field.type() instanceof DoubleType) {
+ rowValues.add(random.nextDouble());
+ } else if (field.type() instanceof VarCharType) {
+ rowValues.add(BinaryString.fromString(field.name() + "_" + rowIdx + "_" + i));
+ } else if (field.type() instanceof LocalZonedTimestampType) {
+ rowValues.add(Timestamp.fromEpochMillis(System.currentTimeMillis()));
+ } else if (field.type() instanceof BooleanType) {
+ rowValues.add(random.nextBoolean());
+ } else {
+ throw new UnsupportedOperationException("Unsupported field type: " + field.type());
+ }
+ }
+
+ return GenericRow.of(rowValues.toArray());
+ }
+
+ private static String initBasePath(Path tempDir, String tableName) {
+ try {
+ Path basePath = tempDir.resolve(tableName);
+ Files.createDirectories(basePath);
+ return basePath.toUri().toString();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public List<GenericRow> insertRows(int numRows) {
+ String partitionValue = LEVEL_VALUES.get(0);
+ return insertRecordsToPartition(numRows, partitionValue);
+ }
+
+ @Override
+ public List<GenericRow> insertRecordsForSpecialPartition(int numRows) {
+ return insertRecordsToPartition(numRows, SPECIAL_PARTITION_VALUE);
+ }
+
+ private List<GenericRow> insertRecordsToPartition(int numRows, String partitionValue) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ List<GenericRow> rows = new ArrayList<>(numRows);
+ for (int i = 0; i < numRows; i++) {
+ GenericRow row = buildGenericRow(i, paimonTable.schema(), partitionValue);
+ writer.write(row);
+ rows.add(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ return rows;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to insert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void upsertRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to upsert rows into Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteRows(List<GenericRow> rows) {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (GenericRow row : rows) {
+ row.setRowKind(RowKind.DELETE);
+ writer.write(row);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ compactTable();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete rows from Paimon table", e);
+ }
+ }
+
+ private void compactTable() {
+ BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
+ SnapshotReader snapshotReader = paimonTable.newSnapshotReader();
+ try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
+ for (BucketEntry bucketEntry : snapshotReader.bucketEntries()) {
+ writer.compact(bucketEntry.partition(), bucketEntry.bucket(), true);
+ }
+ commitWrites(batchWriteBuilder, writer);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to compact writes in Paimon table", e);
+ }
+ }
+
+ private static void commitWrites(BatchWriteBuilder batchWriteBuilder, BatchTableWrite writer)
+ throws Exception {
+ BatchTableCommit commit = batchWriteBuilder.newCommit();
+ List<CommitMessage> messages = writer.prepareCommit();
+ try {
+ commit.commit(messages);
+ } catch (Exception e) {
+ commit.abort(messages);
+ throw new RuntimeException("Failed to commit writes to Paimon table", e);
+ } finally {
+ commit.close();
+ }
+ }
+
+ @Override
+ public void deletePartition(String partitionValue) {
+ try (BatchTableCommit commit = paimonTable.newBatchWriteBuilder().newCommit()) {
+ commit.truncatePartitions(
+ ParameterUtils.getPartitions(partitionField + "=" + partitionValue));
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to delete partition from Paimon table", e);
+ }
+ }
+
+ @Override
+ public void deleteSpecialPartition() {
+ deletePartition(SPECIAL_PARTITION_VALUE);
+ }
+
+ @Override
+ public String getBasePath() {
+ return paimonTable.location().toString();
+ }
+
+ @Override
+ public String getMetadataPath() {
+ return paimonTable.snapshotManager().snapshotDirectory().toString();
+ }
+
+ @Override
+ public String getOrderByColumn() {
+ return "id";
+ }
+
+ @Override
+ public void close() {}
+
+ @Override
+ public void reload() {}
+
+ @Override
+ public List<String> getColumnsToSelect() {
+ return paimonTable.schema().fieldNames().stream()
+ .filter(
+ // TODO Hudi thinks that paimon buckets are partition values, not sure how to handle it
+ // filtering out the partition field on the comparison for now
+ field -> !field.equals(partitionField))
Review Comment:
I will play around with this and see how far I can get.
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2415222613
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
Great, I'll do it and let you know!
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2331953688
##########
xtable-core/src/test/java/org/apache/xtable/ITConversionController.java:
##########
@@ -37,16 +35,7 @@
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Base64;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
-import java.util.UUID;
+import java.util.*;
Review Comment:
got it, I'll fix my IntelliJ config!
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "mikedias (via GitHub)" <gi...@apache.org>.
mikedias commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2335299927
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
Mine doesn't not
<img width="1382" height="850" alt="image" src="https://github.com/user-attachments/assets/3c92148c-2aa3-464c-bf7d-364221becbe5" />
so I guess Paimon is not adding the field id? If that's required for iceberg, I wonder how the [paimon -> iceberg](https://paimon.apache.org/docs/master/iceberg/overview/) integration can work... 🤔
maybe something related on how we read iceberg on our test suite? idk...
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2333606911
##########
xtable-core/pom.xml:
##########
@@ -206,6 +218,14 @@
<skip>false</skip>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>11</source>
+ <target>11</target>
Review Comment:
Yes we use Java 11 as the source and when [developing](https://github.com/apache/incubator-xtable?tab=readme-ov-file#building-the-project-and-running-tests) but not our jar [target](https://github.com/apache/incubator-xtable/blob/main/pom.xml#L63) when publishing.
--
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] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2486674646
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonPartitionExtractor.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.paimon;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
+
+import org.apache.xtable.model.schema.InternalField;
+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;
+
+/** Extracts partition spec for Paimon as identity transforms on partition keys. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonPartitionExtractor {
+
+ private final PaimonSchemaExtractor paimonSchemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private static final PaimonPartitionExtractor INSTANCE = new PaimonPartitionExtractor();
+
+ public static PaimonPartitionExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalPartitionField> toInternalPartitionFields(
+ List<String> partitionKeys, InternalSchema schema) {
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return partitionKeys.stream()
+ .map(key -> toPartitionField(key, schema))
+ .collect(Collectors.toList());
+ }
+
+ public List<PartitionValue> toPartitionValues(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ InternalSchema internalSchema = paimonSchemaExtractor.toInternalSchema(table.schema());
+
+ List<PartitionValue> partitionValues = new ArrayList<>();
+ for (Map.Entry<String, String> entry :
+ partitionComputer.generatePartValues(partition).entrySet()) {
+ PartitionValue partitionValue =
+ PartitionValue.builder()
+ .partitionField(toPartitionField(entry.getKey(), internalSchema))
+ .range(Range.scalar(entry.getValue()))
+ .build();
+ partitionValues.add(partitionValue);
+ }
+ return partitionValues;
+ }
+
+ public Optional<String> toPartitionPath(FileStoreTable table, BinaryRow partition) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ return partitionComputer.generatePartValues(partition).entrySet().stream()
+ .map(e -> e.getKey() + "=" + e.getValue())
+ .reduce((a, b) -> a + "/" + b);
+ }
+
+ private InternalPartitionField toPartitionField(String key, InternalSchema schema) {
+ InternalField sourceField =
+ findField(schema, key)
+ .orElseThrow(
+ () -> new IllegalArgumentException("Partition key not found in schema: " + key));
+ return InternalPartitionField.builder()
+ .sourceField(sourceField)
+ .transformType(PartitionTransformType.VALUE)
Review Comment:
There is no documentation at that level for XTable but it is so we can map to the partition types in other systems like Iceberg where you need to specify the [transformation](https://iceberg.apache.org/spec/#partition-transforms) type used. This allows us to properly translate the metadata and allows for partition pruning in the query engines.
For the ranges themselves, the [java docs](https://github.com/apache/incubator-xtable/blob/main/xtable-api/src/main/java/org/apache/xtable/model/stat/Range.java#L43) describe the expectations on the values so that we can also ensure each table format interprets the values correctly.
--
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] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2484844976
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonPartitionExtractor.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.InternalRowPartitionComputer;
+
+import org.apache.xtable.exception.ReadException;
+import org.apache.xtable.model.schema.InternalField;
+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;
+
+/** Extracts partition spec for Paimon as identity transforms on partition keys. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonPartitionExtractor {
+
+ private static final PaimonPartitionExtractor INSTANCE = new PaimonPartitionExtractor();
+
+ public static PaimonPartitionExtractor getInstance() {
+ return INSTANCE;
+ }
+
+ public List<InternalPartitionField> toInternalPartitionFields(
+ List<String> partitionKeys, InternalSchema schema) {
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return partitionKeys.stream()
+ .map(key -> toPartitionField(key, schema))
+ .collect(Collectors.toList());
+ }
+
+ public List<PartitionValue> toPartitionValues(
+ FileStoreTable table, BinaryRow partition, InternalSchema internalSchema) {
+ InternalRowPartitionComputer partitionComputer = newPartitionComputer(table);
+ LinkedHashMap<String, String> partValues = partitionComputer.generatePartValues(partition);
Review Comment:
Nitpick: just use the interface here `Map<String, String>` unless we rely on a particular detail of the LinkedHashMap implementation
##########
xtable-core/src/test/java/org/apache/xtable/paimon/TestPaimonSchemaExtractor.java:
##########
@@ -0,0 +1,528 @@
+/*
+ * 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.paimon;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.CharType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DateType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.FloatType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.SmallIntType;
+import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.types.TinyIntType;
+import org.apache.paimon.types.VarCharType;
+import org.junit.jupiter.api.Test;
+
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+
+public class TestPaimonSchemaExtractor {
+ private static final PaimonSchemaExtractor schemaExtractor = PaimonSchemaExtractor.getInstance();
+
+ private void assertField(DataField paimonField, InternalField expectedInternalField) {
+ assertField(paimonField, expectedInternalField, Collections.emptyList());
+ }
+
+ private void assertField(
+ DataField paimonField, InternalField expectedInternalField, List<String> primaryKeys) {
+ TableSchema paimonSchema =
+ new TableSchema(
+ 0,
+ Collections.singletonList(paimonField),
+ 0,
+ Collections.emptyList(),
+ primaryKeys,
+ new HashMap<>(),
+ "");
+ InternalSchema internalSchema = schemaExtractor.toInternalSchema(paimonSchema);
+ List<InternalField> recordKeyFields =
+ primaryKeys.isEmpty()
+ ? Collections.emptyList()
+ : Collections.singletonList(expectedInternalField);
+ InternalSchema expectedSchema =
+ InternalSchema.builder()
+ .name("record")
+ .dataType(InternalType.RECORD)
+ .fields(Collections.singletonList(expectedInternalField))
+ .recordKeyFields(recordKeyFields)
+ .build();
+ assertEquals(expectedSchema, internalSchema);
+ }
+
+ @Test
+ void testCharField() {
+ DataField paimonField = new DataField(0, "char_field", new CharType(10));
+ InternalField expectedField =
+ InternalField.builder()
+ .name("char_field")
+ .fieldId(0)
+ .schema(
+ InternalSchema.builder()
+ .name("CHAR(10)")
+ .dataType(InternalType.STRING)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testVarcharField() {
+ DataField paimonField = new DataField(1, "varchar_field", new VarCharType(255));
+ InternalField expectedField =
+ InternalField.builder()
+ .name("varchar_field")
+ .fieldId(1)
+ .schema(
+ InternalSchema.builder()
+ .name("VARCHAR(255)")
+ .dataType(InternalType.STRING)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testBooleanField() {
+ DataField paimonField = new DataField(2, "boolean_field", new BooleanType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("boolean_field")
+ .fieldId(2)
+ .schema(
+ InternalSchema.builder()
+ .name("BOOLEAN")
+ .dataType(InternalType.BOOLEAN)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testTinyIntField() {
+ DataField paimonField = new DataField(3, "tinyint_field", new TinyIntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("tinyint_field")
+ .fieldId(3)
+ .schema(
+ InternalSchema.builder()
+ .name("TINYINT")
+ .dataType(InternalType.INT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testSmallIntField() {
+ DataField paimonField = new DataField(4, "smallint_field", new SmallIntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("smallint_field")
+ .fieldId(4)
+ .schema(
+ InternalSchema.builder()
+ .name("SMALLINT")
+ .dataType(InternalType.INT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testIntField() {
+ DataField paimonField = new DataField(5, "int_field", new IntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("int_field")
+ .fieldId(5)
+ .schema(
+ InternalSchema.builder()
+ .name("INT")
+ .dataType(InternalType.INT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testBigIntField() {
+ DataField paimonField = new DataField(6, "bigint_field", new BigIntType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("bigint_field")
+ .fieldId(6)
+ .schema(
+ InternalSchema.builder()
+ .name("BIGINT")
+ .dataType(InternalType.LONG)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testFloatField() {
+ DataField paimonField = new DataField(7, "float_field", new FloatType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("float_field")
+ .fieldId(7)
+ .schema(
+ InternalSchema.builder()
+ .name("FLOAT")
+ .dataType(InternalType.FLOAT)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testDoubleField() {
+ DataField paimonField = new DataField(8, "double_field", new DoubleType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("double_field")
+ .fieldId(8)
+ .schema(
+ InternalSchema.builder()
+ .name("DOUBLE")
+ .dataType(InternalType.DOUBLE)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testDateField() {
+ DataField paimonField = new DataField(9, "date_field", new DateType());
+ InternalField expectedField =
+ InternalField.builder()
+ .name("date_field")
+ .fieldId(9)
+ .schema(
+ InternalSchema.builder()
+ .name("DATE")
+ .dataType(InternalType.DATE)
+ .isNullable(true)
+ .build())
+ .defaultValue(InternalField.Constants.NULL_DEFAULT_VALUE)
+ .build();
+ assertField(paimonField, expectedField);
+ }
+
+ @Test
+ void testTimestampField() {
Review Comment:
Can you add cases for the other supported timestamp precisions? Maybe make it a parameterized test case?
For the precisions that are not supported in XTable, we can throw an exception for now.
--
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] [Draft] Paimon Source Support [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #742:
URL: https://github.com/apache/incubator-xtable/pull/742#discussion_r2414798910
##########
xtable-core/src/main/java/org/apache/xtable/paimon/PaimonSchemaExtractor.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.paimon;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.types.*;
+
+import org.apache.xtable.exception.NotSupportedException;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.schema.SchemaUtils;
+
+/** Converts Paimon RowType to XTable InternalSchema. */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PaimonSchemaExtractor {
Review Comment:
I have merged that PR, please update your branch to pick up the changes when you get a chance.
--
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