You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/11/22 05:58:16 UTC
[GitHub] [flink-table-store] zjureel opened a new pull request, #396: [FLINK-27846] Schema evolution for reading data file
zjureel opened a new pull request, #396:
URL: https://github.com/apache/flink-table-store/pull/396
Currently, the table store uses the latest schema id to read the data file meta. When the schema evolves, it will cause errors, for example:
the schema of underlying data is [1->a, 2->b, 3->c, 4->d] and schema id is 0, where 1/2/3/4 is field id and a/b/c/d is field name
After schema evolution, schema id is 1, and the new schema is [1->a, 3->c, 5->f, 6->b, 7->g]
When table store reads underlying data file, it should use schema 0 with should [1->a, 2->b, 3->c, 4->d], and mapping schema 1 to 0 according to their field ids.
This PR will read the data according to the schema id from the avro/orc/parquet data file, then create index mapping from the table schema and the underlying data schema, so that the table store can read the correct row data through its latest schema.
The main codes are as follows:
1. Added method `valueFields` in `KeyValueFieldsExtractor` to extract fields from `TableSchema`
2. Added `AbstractFileRecordIterator` for `KeyValueDataFileRecordIterator` and `RowDataFileRecordIterator` to create projected row data from table schema to underlying row data
3. Added methods in `SchemaEvolutionUtil` to create index mapping between schemas, convert projection from table to underlying data
4. Added `BulkFormatMapping` to create reader factory and index mapping for `KeyValueFileReaderFactory` and `AppendOnlyFileStoreRead`
The main tests include:
1. Updated `SchemaEvolutionUtilTest` to create index mapping and convert projection
2. Added `AppendOnlyFileDataTableTest`, `ChangelogValueCountFileDataTableTest` and `ChangelogWithKeyFileDataTableTest` to read and filter data after schema evolution
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #396: [FLINK-27846] Schema evolution for reading data file
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #396:
URL: https://github.com/apache/flink-table-store/pull/396#discussion_r1031306773
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/BulkFormatMapping.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * 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.flink.table.store.file.utils;
+
+import org.apache.flink.connector.file.src.FileSourceSplit;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.store.file.KeyValue;
+import org.apache.flink.table.store.file.predicate.Predicate;
+import org.apache.flink.table.store.file.schema.DataField;
+import org.apache.flink.table.store.file.schema.KeyValueFieldsExtractor;
+import org.apache.flink.table.store.file.schema.RowDataType;
+import org.apache.flink.table.store.file.schema.SchemaEvolutionUtil;
+import org.apache.flink.table.store.file.schema.TableSchema;
+import org.apache.flink.table.store.format.FileFormat;
+import org.apache.flink.table.store.utils.Projection;
+import org.apache.flink.table.types.logical.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/** Class with index mapping and bulk format. */
+public class BulkFormatMapping {
+ @Nullable private final int[] indexMapping;
+ private final BulkFormat<RowData, FileSourceSplit> bulkFormat;
+
+ public BulkFormatMapping(int[] indexMapping, BulkFormat<RowData, FileSourceSplit> bulkFormat) {
+ this.indexMapping = indexMapping;
+ this.bulkFormat = bulkFormat;
+ }
+
+ @Nullable
+ public int[] getIndexMapping() {
+ return indexMapping;
+ }
+
+ public BulkFormat<RowData, FileSourceSplit> getReaderFactory() {
+ return bulkFormat;
+ }
+
+ public static BulkFormatMappingBuilder newBuilder(
+ FileFormat fileFormat,
+ KeyValueFieldsExtractor extractor,
+ int[][] keyProjection,
+ int[][] valueProjection,
+ int[][] projection,
+ @Nullable List<Predicate> filters) {
+ return new BulkFormatMappingBuilder(
+ fileFormat, extractor, keyProjection, valueProjection, projection, filters);
+ }
+
+ /** Builder to build {@link BulkFormatMapping}. */
+ public static class BulkFormatMappingBuilder {
+ private final FileFormat fileFormat;
+ private final KeyValueFieldsExtractor extractor;
+ private final int[][] keyProjection;
+ private final int[][] valueProjection;
+ private final int[][] projection;
+ @Nullable private final List<Predicate> filters;
+
+ private BulkFormatMappingBuilder(
+ FileFormat fileFormat,
+ KeyValueFieldsExtractor extractor,
+ int[][] keyProjection,
+ int[][] valueProjection,
+ int[][] projection,
+ @Nullable List<Predicate> filters) {
+ this.fileFormat = fileFormat;
+ this.extractor = extractor;
+ this.keyProjection = keyProjection;
+ this.valueProjection = valueProjection;
+ this.projection = projection;
+ this.filters = filters;
+ }
+
+ public BulkFormatMapping build(TableSchema tableSchema, TableSchema dataSchema) {
Review Comment:
Can we simplify this process? For example, we can build new special KeyValue fieldIds based on all fields of KeyValue, so that this process can be similar to Append only, and should be simplified.
What do you think?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi merged pull request #396: [FLINK-27846] Schema evolution for reading data file
Posted by GitBox <gi...@apache.org>.
JingsongLi merged PR #396:
URL: https://github.com/apache/flink-table-store/pull/396
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #396: [FLINK-27846] Schema evolution for reading data file
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #396:
URL: https://github.com/apache/flink-table-store/pull/396#discussion_r1031148421
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaEvolutionUtil.java:
##########
@@ -62,4 +72,193 @@ public static int[] createIndexMapping(
}
return null;
}
+
+ /**
+ * Create index mapping from table projection to underlying data projection.
+ *
+ * @param tableProjection the table projection
+ * @param tableFields the fields in table
+ * @param dataProjection the underlying data projection
+ * @param dataFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ List<DataField> tableFields,
+ int[] dataProjection,
+ List<DataField> dataFields) {
+ return createIndexMapping(
+ tableProjection,
+ tableProjection.length,
+ tableFields,
+ Collections.emptyList(),
+ dataProjection,
+ dataProjection.length,
+ dataFields,
+ Collections.emptyList());
+ }
+
+ /**
+ * Create index mapping from table projection to underlying data projection for key value.
+ *
+ * @param tableProjection the table projection
+ * @param tableKeyCount the key count in table
+ * @param tableKeyFields the key fields in table
+ * @param tableValueFields the value fields in table
+ * @param dataProjection the data projection
+ * @param dataKeyCount the data key count
+ * @param dataKeyFields the fields in underlying data
+ * @param dataValueFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ int tableKeyCount,
+ List<DataField> tableKeyFields,
+ List<DataField> tableValueFields,
+ int[] dataProjection,
+ int dataKeyCount,
+ List<DataField> dataKeyFields,
+ List<DataField> dataValueFields) {
+ List<Integer> tableKeyFieldIdList =
+ tableKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataKeyFieldIdList =
+ dataKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ int[] indexMapping = new int[tableProjection.length];
+
+ int[] dataKeyProjection = Arrays.copyOf(dataProjection, dataKeyCount);
+ for (int i = 0; i < tableKeyCount; i++) {
+ int fieldId = tableKeyFieldIdList.get(tableProjection[i]);
+ int dataFieldIndex = dataKeyFieldIdList.indexOf(fieldId);
+ indexMapping[i] = Ints.indexOf(dataKeyProjection, dataFieldIndex);
+ }
+ if (tableProjection.length >= tableKeyCount + 2) {
+ // seq and value kind
+ for (int i = tableKeyCount; i < tableKeyCount + 2; i++) {
+ indexMapping[i] = i + dataKeyCount - tableKeyCount;
+ }
+
+ int[] dataValueProjection =
+ Arrays.copyOfRange(dataProjection, dataKeyCount + 2, dataProjection.length);
+ for (int i = 0; i < dataValueProjection.length; i++) {
+ dataValueProjection[i] = dataValueProjection[i] - dataKeyFields.size() - 2;
+ }
+ List<Integer> tableValueFieldIdList =
+ tableValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataValueFieldIdList =
+ dataValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ for (int i = tableKeyCount + 2; i < tableProjection.length; i++) {
+ int fieldId =
+ tableValueFieldIdList.get(tableProjection[i] - tableKeyFields.size() - 2);
+ int dataFieldIndex = dataValueFieldIdList.indexOf(fieldId);
+ int dataValueIndex = Ints.indexOf(dataValueProjection, dataFieldIndex);
+ indexMapping[i] =
+ dataValueIndex < 0 ? dataValueIndex : dataValueIndex + dataKeyCount + 2;
+ }
+ }
+
+ for (int i = 0; i < indexMapping.length; i++) {
+ if (indexMapping[i] != i) {
+ return indexMapping;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Create data projection from table projection.
+ *
+ * @param tableFields the fields of table
+ * @param dataFields the fields of underlying data
+ * @param tableProjection the projection of table
+ * @return the projection of data
+ */
+ public static int[][] createDataProjection(
+ List<DataField> tableFields, List<DataField> dataFields, int[][] tableProjection) {
+ List<Integer> tableFieldIdList =
Review Comment:
Here we can validate `tableProjection` can not be nested projection.
`Projection.of(tableProjection).toTopLevelIndexes()` to get top level indexes.
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/KeyValueDataFileRecordReader.java:
##########
@@ -80,7 +85,7 @@ public KeyValue next() throws IOException {
if (result == null) {
return null;
} else {
- return serializer.fromRow(result.getRecord()).setLevel(level);
+ return serializer.fromRow(mappingRowData(result.getRecord())).setLevel(level);
Review Comment:
You can remove above comments: `TODO schema evolution`
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaEvolutionUtil.java:
##########
@@ -62,4 +72,193 @@ public static int[] createIndexMapping(
}
return null;
}
+
+ /**
+ * Create index mapping from table projection to underlying data projection.
+ *
+ * @param tableProjection the table projection
+ * @param tableFields the fields in table
+ * @param dataProjection the underlying data projection
+ * @param dataFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ List<DataField> tableFields,
+ int[] dataProjection,
+ List<DataField> dataFields) {
+ return createIndexMapping(
+ tableProjection,
+ tableProjection.length,
+ tableFields,
+ Collections.emptyList(),
+ dataProjection,
+ dataProjection.length,
+ dataFields,
+ Collections.emptyList());
+ }
+
+ /**
+ * Create index mapping from table projection to underlying data projection for key value.
+ *
+ * @param tableProjection the table projection
+ * @param tableKeyCount the key count in table
+ * @param tableKeyFields the key fields in table
+ * @param tableValueFields the value fields in table
+ * @param dataProjection the data projection
+ * @param dataKeyCount the data key count
+ * @param dataKeyFields the fields in underlying data
+ * @param dataValueFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ int tableKeyCount,
+ List<DataField> tableKeyFields,
+ List<DataField> tableValueFields,
+ int[] dataProjection,
+ int dataKeyCount,
+ List<DataField> dataKeyFields,
+ List<DataField> dataValueFields) {
+ List<Integer> tableKeyFieldIdList =
+ tableKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataKeyFieldIdList =
+ dataKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ int[] indexMapping = new int[tableProjection.length];
+
+ int[] dataKeyProjection = Arrays.copyOf(dataProjection, dataKeyCount);
+ for (int i = 0; i < tableKeyCount; i++) {
+ int fieldId = tableKeyFieldIdList.get(tableProjection[i]);
+ int dataFieldIndex = dataKeyFieldIdList.indexOf(fieldId);
+ indexMapping[i] = Ints.indexOf(dataKeyProjection, dataFieldIndex);
+ }
+ if (tableProjection.length >= tableKeyCount + 2) {
+ // seq and value kind
+ for (int i = tableKeyCount; i < tableKeyCount + 2; i++) {
+ indexMapping[i] = i + dataKeyCount - tableKeyCount;
+ }
+
+ int[] dataValueProjection =
+ Arrays.copyOfRange(dataProjection, dataKeyCount + 2, dataProjection.length);
+ for (int i = 0; i < dataValueProjection.length; i++) {
+ dataValueProjection[i] = dataValueProjection[i] - dataKeyFields.size() - 2;
+ }
+ List<Integer> tableValueFieldIdList =
+ tableValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataValueFieldIdList =
+ dataValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ for (int i = tableKeyCount + 2; i < tableProjection.length; i++) {
+ int fieldId =
+ tableValueFieldIdList.get(tableProjection[i] - tableKeyFields.size() - 2);
+ int dataFieldIndex = dataValueFieldIdList.indexOf(fieldId);
+ int dataValueIndex = Ints.indexOf(dataValueProjection, dataFieldIndex);
+ indexMapping[i] =
+ dataValueIndex < 0 ? dataValueIndex : dataValueIndex + dataKeyCount + 2;
+ }
+ }
+
+ for (int i = 0; i < indexMapping.length; i++) {
+ if (indexMapping[i] != i) {
+ return indexMapping;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Create data projection from table projection.
+ *
+ * @param tableFields the fields of table
+ * @param dataFields the fields of underlying data
+ * @param tableProjection the projection of table
+ * @return the projection of data
+ */
+ public static int[][] createDataProjection(
+ List<DataField> tableFields, List<DataField> dataFields, int[][] tableProjection) {
+ List<Integer> tableFieldIdList =
+ tableFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataFieldIdList =
+ dataFields.stream().map(DataField::id).collect(Collectors.toList());
+ return Arrays.stream(tableProjection)
+ .map(p -> Arrays.copyOf(p, p.length))
+ .peek(
+ p -> {
+ int fieldId = tableFieldIdList.get(p[0]);
+ p[0] = dataFieldIdList.indexOf(fieldId);
+ })
+ .filter(p -> p[0] >= 0)
+ .toArray(int[][]::new);
+ }
+
+ /**
+ * Create predicate list from data fields.
+ *
+ * @param tableFields the table fields
+ * @param dataFields the underlying data fields
+ * @param filters the filters
+ * @return the data filters
+ */
+ public static List<Predicate> createDataFilters(
+ List<DataField> tableFields, List<DataField> dataFields, List<Predicate> filters) {
+ if (filters == null) {
+ return null;
+ }
+
+ List<Predicate> dataFilters = new ArrayList<>(filters.size());
+ for (Predicate predicate : filters) {
+ dataFilters.add(createDataPredicate(tableFields, dataFields, predicate));
+ }
+ return dataFilters;
+ }
+
+ @Nullable
+ private static Predicate createDataPredicate(
+ List<DataField> tableFields, List<DataField> dataFields, Predicate predicate) {
+ if (predicate instanceof CompoundPredicate) {
+ CompoundPredicate compoundPredicate = (CompoundPredicate) predicate;
+ List<Predicate> children = compoundPredicate.children();
+ List<Predicate> dataChildren = new ArrayList<>(children.size());
+ for (Predicate child : children) {
+ Predicate dataPredicate = createDataPredicate(tableFields, dataFields, child);
+ if (dataPredicate != null) {
+ dataChildren.add(dataPredicate);
+ }
+ }
+ return new CompoundPredicate(compoundPredicate.function(), dataChildren);
+ } else if (predicate instanceof LeafPredicate) {
+ LeafPredicate leafPredicate = (LeafPredicate) predicate;
+ List<DataField> predicateTableFields =
+ tableFields.stream()
+ .filter(f -> f.name().equals(leafPredicate.fieldName()))
+ .collect(Collectors.toList());
+ if (predicateTableFields.size() != 1) {
+ throw new IllegalArgumentException(
+ String.format("Find none or multiple fields %s", predicateTableFields));
+ }
+ DataField tableField = predicateTableFields.get(0);
+ List<DataField> predicateDataFields =
+ dataFields.stream()
+ .filter(f -> f.id() == tableField.id())
+ .collect(Collectors.toList());
+ if (predicateDataFields.isEmpty()) {
+ return null;
+ } else if (predicateDataFields.size() > 1) {
+ throw new IllegalArgumentException(
+ String.format("Find none or multiple fields %s", predicateTableFields));
+ }
+ DataField dataField = predicateDataFields.get(0);
+ return new LeafPredicate(
+ leafPredicate.function(),
+ leafPredicate.type(),
+ dataFields.indexOf(dataField),
+ dataField.name(),
+ leafPredicate.literals());
Review Comment:
There are also troublesome things here. We may need to deal with type promotion (such as int ->bigint). I don't know whether orc will help me with this.
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaEvolutionUtil.java:
##########
@@ -62,4 +72,193 @@ public static int[] createIndexMapping(
}
return null;
}
+
+ /**
+ * Create index mapping from table projection to underlying data projection.
+ *
+ * @param tableProjection the table projection
+ * @param tableFields the fields in table
+ * @param dataProjection the underlying data projection
+ * @param dataFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ List<DataField> tableFields,
+ int[] dataProjection,
+ List<DataField> dataFields) {
+ return createIndexMapping(
+ tableProjection,
+ tableProjection.length,
+ tableFields,
+ Collections.emptyList(),
+ dataProjection,
+ dataProjection.length,
+ dataFields,
+ Collections.emptyList());
+ }
+
+ /**
+ * Create index mapping from table projection to underlying data projection for key value.
+ *
+ * @param tableProjection the table projection
+ * @param tableKeyCount the key count in table
+ * @param tableKeyFields the key fields in table
+ * @param tableValueFields the value fields in table
+ * @param dataProjection the data projection
+ * @param dataKeyCount the data key count
+ * @param dataKeyFields the fields in underlying data
+ * @param dataValueFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ int tableKeyCount,
+ List<DataField> tableKeyFields,
+ List<DataField> tableValueFields,
+ int[] dataProjection,
+ int dataKeyCount,
+ List<DataField> dataKeyFields,
+ List<DataField> dataValueFields) {
+ List<Integer> tableKeyFieldIdList =
+ tableKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataKeyFieldIdList =
+ dataKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ int[] indexMapping = new int[tableProjection.length];
+
+ int[] dataKeyProjection = Arrays.copyOf(dataProjection, dataKeyCount);
+ for (int i = 0; i < tableKeyCount; i++) {
+ int fieldId = tableKeyFieldIdList.get(tableProjection[i]);
+ int dataFieldIndex = dataKeyFieldIdList.indexOf(fieldId);
+ indexMapping[i] = Ints.indexOf(dataKeyProjection, dataFieldIndex);
+ }
+ if (tableProjection.length >= tableKeyCount + 2) {
+ // seq and value kind
+ for (int i = tableKeyCount; i < tableKeyCount + 2; i++) {
+ indexMapping[i] = i + dataKeyCount - tableKeyCount;
+ }
+
+ int[] dataValueProjection =
+ Arrays.copyOfRange(dataProjection, dataKeyCount + 2, dataProjection.length);
+ for (int i = 0; i < dataValueProjection.length; i++) {
+ dataValueProjection[i] = dataValueProjection[i] - dataKeyFields.size() - 2;
+ }
+ List<Integer> tableValueFieldIdList =
+ tableValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataValueFieldIdList =
+ dataValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ for (int i = tableKeyCount + 2; i < tableProjection.length; i++) {
+ int fieldId =
+ tableValueFieldIdList.get(tableProjection[i] - tableKeyFields.size() - 2);
+ int dataFieldIndex = dataValueFieldIdList.indexOf(fieldId);
+ int dataValueIndex = Ints.indexOf(dataValueProjection, dataFieldIndex);
+ indexMapping[i] =
+ dataValueIndex < 0 ? dataValueIndex : dataValueIndex + dataKeyCount + 2;
+ }
+ }
+
+ for (int i = 0; i < indexMapping.length; i++) {
+ if (indexMapping[i] != i) {
+ return indexMapping;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Create data projection from table projection.
+ *
+ * @param tableFields the fields of table
+ * @param dataFields the fields of underlying data
+ * @param tableProjection the projection of table
+ * @return the projection of data
+ */
+ public static int[][] createDataProjection(
+ List<DataField> tableFields, List<DataField> dataFields, int[][] tableProjection) {
+ List<Integer> tableFieldIdList =
+ tableFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataFieldIdList =
+ dataFields.stream().map(DataField::id).collect(Collectors.toList());
+ return Arrays.stream(tableProjection)
+ .map(p -> Arrays.copyOf(p, p.length))
+ .peek(
+ p -> {
+ int fieldId = tableFieldIdList.get(p[0]);
+ p[0] = dataFieldIdList.indexOf(fieldId);
+ })
+ .filter(p -> p[0] >= 0)
+ .toArray(int[][]::new);
+ }
+
+ /**
+ * Create predicate list from data fields.
+ *
+ * @param tableFields the table fields
+ * @param dataFields the underlying data fields
+ * @param filters the filters
+ * @return the data filters
+ */
+ public static List<Predicate> createDataFilters(
+ List<DataField> tableFields, List<DataField> dataFields, List<Predicate> filters) {
+ if (filters == null) {
+ return null;
+ }
+
+ List<Predicate> dataFilters = new ArrayList<>(filters.size());
+ for (Predicate predicate : filters) {
+ dataFilters.add(createDataPredicate(tableFields, dataFields, predicate));
+ }
+ return dataFilters;
+ }
+
+ @Nullable
+ private static Predicate createDataPredicate(
+ List<DataField> tableFields, List<DataField> dataFields, Predicate predicate) {
+ if (predicate instanceof CompoundPredicate) {
+ CompoundPredicate compoundPredicate = (CompoundPredicate) predicate;
+ List<Predicate> children = compoundPredicate.children();
+ List<Predicate> dataChildren = new ArrayList<>(children.size());
+ for (Predicate child : children) {
+ Predicate dataPredicate = createDataPredicate(tableFields, dataFields, child);
+ if (dataPredicate != null) {
+ dataChildren.add(dataPredicate);
+ }
+ }
+ return new CompoundPredicate(compoundPredicate.function(), dataChildren);
Review Comment:
We could think about predicates deeply.
Not found means the column is null, not means we can skip this predicate?
For example:
`f0 = 1 or f1 is null`. f1 is not found in the data file. This means we can not just return a `f0 = 1`, this will cause us to filter out the required data.
We can add cases for this.
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaEvolutionUtil.java:
##########
@@ -62,4 +72,193 @@ public static int[] createIndexMapping(
}
return null;
}
+
+ /**
+ * Create index mapping from table projection to underlying data projection.
+ *
+ * @param tableProjection the table projection
+ * @param tableFields the fields in table
+ * @param dataProjection the underlying data projection
+ * @param dataFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ List<DataField> tableFields,
+ int[] dataProjection,
+ List<DataField> dataFields) {
+ return createIndexMapping(
+ tableProjection,
+ tableProjection.length,
+ tableFields,
+ Collections.emptyList(),
+ dataProjection,
+ dataProjection.length,
+ dataFields,
+ Collections.emptyList());
+ }
+
+ /**
+ * Create index mapping from table projection to underlying data projection for key value.
+ *
+ * @param tableProjection the table projection
+ * @param tableKeyCount the key count in table
+ * @param tableKeyFields the key fields in table
+ * @param tableValueFields the value fields in table
+ * @param dataProjection the data projection
+ * @param dataKeyCount the data key count
+ * @param dataKeyFields the fields in underlying data
+ * @param dataValueFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ int tableKeyCount,
+ List<DataField> tableKeyFields,
+ List<DataField> tableValueFields,
+ int[] dataProjection,
+ int dataKeyCount,
+ List<DataField> dataKeyFields,
+ List<DataField> dataValueFields) {
+ List<Integer> tableKeyFieldIdList =
+ tableKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataKeyFieldIdList =
+ dataKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ int[] indexMapping = new int[tableProjection.length];
+
+ int[] dataKeyProjection = Arrays.copyOf(dataProjection, dataKeyCount);
+ for (int i = 0; i < tableKeyCount; i++) {
+ int fieldId = tableKeyFieldIdList.get(tableProjection[i]);
+ int dataFieldIndex = dataKeyFieldIdList.indexOf(fieldId);
+ indexMapping[i] = Ints.indexOf(dataKeyProjection, dataFieldIndex);
+ }
+ if (tableProjection.length >= tableKeyCount + 2) {
+ // seq and value kind
+ for (int i = tableKeyCount; i < tableKeyCount + 2; i++) {
+ indexMapping[i] = i + dataKeyCount - tableKeyCount;
+ }
+
+ int[] dataValueProjection =
+ Arrays.copyOfRange(dataProjection, dataKeyCount + 2, dataProjection.length);
+ for (int i = 0; i < dataValueProjection.length; i++) {
+ dataValueProjection[i] = dataValueProjection[i] - dataKeyFields.size() - 2;
+ }
+ List<Integer> tableValueFieldIdList =
+ tableValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataValueFieldIdList =
+ dataValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ for (int i = tableKeyCount + 2; i < tableProjection.length; i++) {
+ int fieldId =
+ tableValueFieldIdList.get(tableProjection[i] - tableKeyFields.size() - 2);
+ int dataFieldIndex = dataValueFieldIdList.indexOf(fieldId);
+ int dataValueIndex = Ints.indexOf(dataValueProjection, dataFieldIndex);
+ indexMapping[i] =
+ dataValueIndex < 0 ? dataValueIndex : dataValueIndex + dataKeyCount + 2;
+ }
+ }
+
+ for (int i = 0; i < indexMapping.length; i++) {
+ if (indexMapping[i] != i) {
+ return indexMapping;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Create data projection from table projection.
+ *
+ * @param tableFields the fields of table
+ * @param dataFields the fields of underlying data
+ * @param tableProjection the projection of table
+ * @return the projection of data
+ */
+ public static int[][] createDataProjection(
+ List<DataField> tableFields, List<DataField> dataFields, int[][] tableProjection) {
+ List<Integer> tableFieldIdList =
+ tableFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataFieldIdList =
+ dataFields.stream().map(DataField::id).collect(Collectors.toList());
+ return Arrays.stream(tableProjection)
+ .map(p -> Arrays.copyOf(p, p.length))
+ .peek(
+ p -> {
+ int fieldId = tableFieldIdList.get(p[0]);
+ p[0] = dataFieldIdList.indexOf(fieldId);
+ })
+ .filter(p -> p[0] >= 0)
+ .toArray(int[][]::new);
+ }
+
+ /**
+ * Create predicate list from data fields.
+ *
+ * @param tableFields the table fields
+ * @param dataFields the underlying data fields
+ * @param filters the filters
+ * @return the data filters
+ */
+ public static List<Predicate> createDataFilters(
+ List<DataField> tableFields, List<DataField> dataFields, List<Predicate> filters) {
+ if (filters == null) {
+ return null;
+ }
+
+ List<Predicate> dataFilters = new ArrayList<>(filters.size());
+ for (Predicate predicate : filters) {
+ dataFilters.add(createDataPredicate(tableFields, dataFields, predicate));
+ }
+ return dataFilters;
+ }
+
+ @Nullable
+ private static Predicate createDataPredicate(
+ List<DataField> tableFields, List<DataField> dataFields, Predicate predicate) {
+ if (predicate instanceof CompoundPredicate) {
+ CompoundPredicate compoundPredicate = (CompoundPredicate) predicate;
+ List<Predicate> children = compoundPredicate.children();
+ List<Predicate> dataChildren = new ArrayList<>(children.size());
+ for (Predicate child : children) {
+ Predicate dataPredicate = createDataPredicate(tableFields, dataFields, child);
+ if (dataPredicate != null) {
+ dataChildren.add(dataPredicate);
+ }
+ }
+ return new CompoundPredicate(compoundPredicate.function(), dataChildren);
+ } else if (predicate instanceof LeafPredicate) {
+ LeafPredicate leafPredicate = (LeafPredicate) predicate;
+ List<DataField> predicateTableFields =
+ tableFields.stream()
+ .filter(f -> f.name().equals(leafPredicate.fieldName()))
+ .collect(Collectors.toList());
+ if (predicateTableFields.size() != 1) {
+ throw new IllegalArgumentException(
+ String.format("Find none or multiple fields %s", predicateTableFields));
+ }
+ DataField tableField = predicateTableFields.get(0);
+ List<DataField> predicateDataFields =
+ dataFields.stream()
+ .filter(f -> f.id() == tableField.id())
+ .collect(Collectors.toList());
+ if (predicateDataFields.isEmpty()) {
+ return null;
+ } else if (predicateDataFields.size() > 1) {
Review Comment:
Maybe here we can create a `Map<fieldId, Field<index, name, type>>` for `dataFields`.
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/BulkFormatMapping.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * 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.flink.table.store.file.utils;
+
+import org.apache.flink.connector.file.src.FileSourceSplit;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.store.file.KeyValue;
+import org.apache.flink.table.store.file.predicate.Predicate;
+import org.apache.flink.table.store.file.schema.DataField;
+import org.apache.flink.table.store.file.schema.KeyValueFieldsExtractor;
+import org.apache.flink.table.store.file.schema.RowDataType;
+import org.apache.flink.table.store.file.schema.SchemaEvolutionUtil;
+import org.apache.flink.table.store.file.schema.TableSchema;
+import org.apache.flink.table.store.format.FileFormat;
+import org.apache.flink.table.store.utils.Projection;
+import org.apache.flink.table.types.logical.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/** Class with index mapping and bulk format. */
+public class BulkFormatMapping {
+ @Nullable private final int[] indexMapping;
+ private final BulkFormat<RowData, FileSourceSplit> bulkFormat;
+
+ public BulkFormatMapping(int[] indexMapping, BulkFormat<RowData, FileSourceSplit> bulkFormat) {
+ this.indexMapping = indexMapping;
+ this.bulkFormat = bulkFormat;
+ }
+
+ @Nullable
+ public int[] getIndexMapping() {
+ return indexMapping;
+ }
+
+ public BulkFormat<RowData, FileSourceSplit> getReaderFactory() {
+ return bulkFormat;
+ }
+
+ public static BulkFormatMappingBuilder newBuilder(
+ FileFormat fileFormat,
+ KeyValueFieldsExtractor extractor,
+ int[][] keyProjection,
+ int[][] valueProjection,
+ int[][] projection,
+ @Nullable List<Predicate> filters) {
+ return new BulkFormatMappingBuilder(
+ fileFormat, extractor, keyProjection, valueProjection, projection, filters);
+ }
+
+ /** Builder to build {@link BulkFormatMapping}. */
+ public static class BulkFormatMappingBuilder {
+ private final FileFormat fileFormat;
+ private final KeyValueFieldsExtractor extractor;
+ private final int[][] keyProjection;
+ private final int[][] valueProjection;
+ private final int[][] projection;
+ @Nullable private final List<Predicate> filters;
+
+ private BulkFormatMappingBuilder(
+ FileFormat fileFormat,
+ KeyValueFieldsExtractor extractor,
+ int[][] keyProjection,
+ int[][] valueProjection,
+ int[][] projection,
+ @Nullable List<Predicate> filters) {
+ this.fileFormat = fileFormat;
+ this.extractor = extractor;
+ this.keyProjection = keyProjection;
+ this.valueProjection = valueProjection;
+ this.projection = projection;
+ this.filters = filters;
+ }
+
+ public BulkFormatMapping build(TableSchema tableSchema, TableSchema dataSchema) {
+ List<DataField> tableKeyFields = extractor.keyFields(tableSchema);
+ List<DataField> tableValueFields = extractor.valueFields(tableSchema);
+
+ List<DataField> dataKeyFields = extractor.keyFields(dataSchema);
+ List<DataField> dataValueFields = extractor.valueFields(dataSchema);
+ int[][] dataKeyProjection =
+ SchemaEvolutionUtil.createDataProjection(
+ tableKeyFields, dataKeyFields, keyProjection);
+ int[][] dataValueProjection =
+ SchemaEvolutionUtil.createDataProjection(
+ tableValueFields, dataValueFields, valueProjection);
+
+ RowType keyType = RowDataType.toRowType(false, dataKeyFields);
+ RowType valueType = RowDataType.toRowType(false, dataValueFields);
+ RowType dataRecordType = KeyValue.schema(keyType, valueType);
+ int[][] dataProjection =
+ KeyValue.project(
+ dataKeyProjection, dataValueProjection, keyType.getFieldCount());
+
+ int[] indexMapping =
+ SchemaEvolutionUtil.createIndexMapping(
Review Comment:
I don't quite understand why we do createDataProjection first and then createIndexMapping. Can you give me an example?
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaEvolutionUtil.java:
##########
@@ -62,4 +72,193 @@ public static int[] createIndexMapping(
}
return null;
}
+
+ /**
+ * Create index mapping from table projection to underlying data projection.
+ *
+ * @param tableProjection the table projection
+ * @param tableFields the fields in table
+ * @param dataProjection the underlying data projection
+ * @param dataFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ List<DataField> tableFields,
+ int[] dataProjection,
+ List<DataField> dataFields) {
+ return createIndexMapping(
+ tableProjection,
+ tableProjection.length,
+ tableFields,
+ Collections.emptyList(),
+ dataProjection,
+ dataProjection.length,
+ dataFields,
+ Collections.emptyList());
+ }
+
+ /**
+ * Create index mapping from table projection to underlying data projection for key value.
+ *
+ * @param tableProjection the table projection
+ * @param tableKeyCount the key count in table
+ * @param tableKeyFields the key fields in table
+ * @param tableValueFields the value fields in table
+ * @param dataProjection the data projection
+ * @param dataKeyCount the data key count
+ * @param dataKeyFields the fields in underlying data
+ * @param dataValueFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ int tableKeyCount,
+ List<DataField> tableKeyFields,
+ List<DataField> tableValueFields,
+ int[] dataProjection,
+ int dataKeyCount,
+ List<DataField> dataKeyFields,
+ List<DataField> dataValueFields) {
+ List<Integer> tableKeyFieldIdList =
+ tableKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataKeyFieldIdList =
+ dataKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ int[] indexMapping = new int[tableProjection.length];
+
+ int[] dataKeyProjection = Arrays.copyOf(dataProjection, dataKeyCount);
+ for (int i = 0; i < tableKeyCount; i++) {
+ int fieldId = tableKeyFieldIdList.get(tableProjection[i]);
+ int dataFieldIndex = dataKeyFieldIdList.indexOf(fieldId);
+ indexMapping[i] = Ints.indexOf(dataKeyProjection, dataFieldIndex);
+ }
+ if (tableProjection.length >= tableKeyCount + 2) {
+ // seq and value kind
+ for (int i = tableKeyCount; i < tableKeyCount + 2; i++) {
+ indexMapping[i] = i + dataKeyCount - tableKeyCount;
+ }
+
+ int[] dataValueProjection =
+ Arrays.copyOfRange(dataProjection, dataKeyCount + 2, dataProjection.length);
+ for (int i = 0; i < dataValueProjection.length; i++) {
+ dataValueProjection[i] = dataValueProjection[i] - dataKeyFields.size() - 2;
+ }
+ List<Integer> tableValueFieldIdList =
+ tableValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataValueFieldIdList =
+ dataValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ for (int i = tableKeyCount + 2; i < tableProjection.length; i++) {
+ int fieldId =
+ tableValueFieldIdList.get(tableProjection[i] - tableKeyFields.size() - 2);
+ int dataFieldIndex = dataValueFieldIdList.indexOf(fieldId);
+ int dataValueIndex = Ints.indexOf(dataValueProjection, dataFieldIndex);
+ indexMapping[i] =
+ dataValueIndex < 0 ? dataValueIndex : dataValueIndex + dataKeyCount + 2;
+ }
+ }
+
+ for (int i = 0; i < indexMapping.length; i++) {
+ if (indexMapping[i] != i) {
+ return indexMapping;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Create data projection from table projection.
+ *
+ * @param tableFields the fields of table
+ * @param dataFields the fields of underlying data
+ * @param tableProjection the projection of table
+ * @return the projection of data
+ */
+ public static int[][] createDataProjection(
+ List<DataField> tableFields, List<DataField> dataFields, int[][] tableProjection) {
+ List<Integer> tableFieldIdList =
+ tableFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataFieldIdList =
+ dataFields.stream().map(DataField::id).collect(Collectors.toList());
+ return Arrays.stream(tableProjection)
+ .map(p -> Arrays.copyOf(p, p.length))
+ .peek(
+ p -> {
+ int fieldId = tableFieldIdList.get(p[0]);
+ p[0] = dataFieldIdList.indexOf(fieldId);
+ })
+ .filter(p -> p[0] >= 0)
Review Comment:
Why filter negative index?
For example, `tableProjection` is {fieldId-0, fieldId-1, fieldId-2}. `fieldId-2` can not be found in data fields.
But we should produce a row with three columns? The third column should be null?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #396: [FLINK-27846] Schema evolution for reading data file
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #396:
URL: https://github.com/apache/flink-table-store/pull/396#discussion_r1031189498
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaEvolutionUtil.java:
##########
@@ -62,4 +72,193 @@ public static int[] createIndexMapping(
}
return null;
}
+
+ /**
+ * Create index mapping from table projection to underlying data projection.
+ *
+ * @param tableProjection the table projection
+ * @param tableFields the fields in table
+ * @param dataProjection the underlying data projection
+ * @param dataFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ List<DataField> tableFields,
+ int[] dataProjection,
+ List<DataField> dataFields) {
+ return createIndexMapping(
+ tableProjection,
+ tableProjection.length,
+ tableFields,
+ Collections.emptyList(),
+ dataProjection,
+ dataProjection.length,
+ dataFields,
+ Collections.emptyList());
+ }
+
+ /**
+ * Create index mapping from table projection to underlying data projection for key value.
+ *
+ * @param tableProjection the table projection
+ * @param tableKeyCount the key count in table
+ * @param tableKeyFields the key fields in table
+ * @param tableValueFields the value fields in table
+ * @param dataProjection the data projection
+ * @param dataKeyCount the data key count
+ * @param dataKeyFields the fields in underlying data
+ * @param dataValueFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ int tableKeyCount,
+ List<DataField> tableKeyFields,
+ List<DataField> tableValueFields,
+ int[] dataProjection,
+ int dataKeyCount,
+ List<DataField> dataKeyFields,
+ List<DataField> dataValueFields) {
+ List<Integer> tableKeyFieldIdList =
+ tableKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataKeyFieldIdList =
+ dataKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ int[] indexMapping = new int[tableProjection.length];
+
+ int[] dataKeyProjection = Arrays.copyOf(dataProjection, dataKeyCount);
+ for (int i = 0; i < tableKeyCount; i++) {
+ int fieldId = tableKeyFieldIdList.get(tableProjection[i]);
+ int dataFieldIndex = dataKeyFieldIdList.indexOf(fieldId);
+ indexMapping[i] = Ints.indexOf(dataKeyProjection, dataFieldIndex);
+ }
+ if (tableProjection.length >= tableKeyCount + 2) {
+ // seq and value kind
+ for (int i = tableKeyCount; i < tableKeyCount + 2; i++) {
+ indexMapping[i] = i + dataKeyCount - tableKeyCount;
+ }
+
+ int[] dataValueProjection =
+ Arrays.copyOfRange(dataProjection, dataKeyCount + 2, dataProjection.length);
+ for (int i = 0; i < dataValueProjection.length; i++) {
+ dataValueProjection[i] = dataValueProjection[i] - dataKeyFields.size() - 2;
+ }
+ List<Integer> tableValueFieldIdList =
+ tableValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataValueFieldIdList =
+ dataValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ for (int i = tableKeyCount + 2; i < tableProjection.length; i++) {
+ int fieldId =
+ tableValueFieldIdList.get(tableProjection[i] - tableKeyFields.size() - 2);
+ int dataFieldIndex = dataValueFieldIdList.indexOf(fieldId);
+ int dataValueIndex = Ints.indexOf(dataValueProjection, dataFieldIndex);
+ indexMapping[i] =
+ dataValueIndex < 0 ? dataValueIndex : dataValueIndex + dataKeyCount + 2;
+ }
+ }
+
+ for (int i = 0; i < indexMapping.length; i++) {
+ if (indexMapping[i] != i) {
+ return indexMapping;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Create data projection from table projection.
+ *
+ * @param tableFields the fields of table
+ * @param dataFields the fields of underlying data
+ * @param tableProjection the projection of table
+ * @return the projection of data
+ */
+ public static int[][] createDataProjection(
+ List<DataField> tableFields, List<DataField> dataFields, int[][] tableProjection) {
+ List<Integer> tableFieldIdList =
+ tableFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataFieldIdList =
+ dataFields.stream().map(DataField::id).collect(Collectors.toList());
+ return Arrays.stream(tableProjection)
+ .map(p -> Arrays.copyOf(p, p.length))
+ .peek(
+ p -> {
+ int fieldId = tableFieldIdList.get(p[0]);
+ p[0] = dataFieldIdList.indexOf(fieldId);
+ })
+ .filter(p -> p[0] >= 0)
Review Comment:
I may have confused it with `createIndexMapping`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #396: [FLINK-27846] Schema evolution for reading data file
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #396:
URL: https://github.com/apache/flink-table-store/pull/396#discussion_r1031267835
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaEvolutionUtil.java:
##########
@@ -62,4 +72,193 @@ public static int[] createIndexMapping(
}
return null;
}
+
+ /**
+ * Create index mapping from table projection to underlying data projection.
+ *
+ * @param tableProjection the table projection
+ * @param tableFields the fields in table
+ * @param dataProjection the underlying data projection
+ * @param dataFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ List<DataField> tableFields,
+ int[] dataProjection,
+ List<DataField> dataFields) {
+ return createIndexMapping(
+ tableProjection,
+ tableProjection.length,
+ tableFields,
+ Collections.emptyList(),
+ dataProjection,
+ dataProjection.length,
+ dataFields,
+ Collections.emptyList());
+ }
+
+ /**
+ * Create index mapping from table projection to underlying data projection for key value.
+ *
+ * @param tableProjection the table projection
+ * @param tableKeyCount the key count in table
+ * @param tableKeyFields the key fields in table
+ * @param tableValueFields the value fields in table
+ * @param dataProjection the data projection
+ * @param dataKeyCount the data key count
+ * @param dataKeyFields the fields in underlying data
+ * @param dataValueFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ int tableKeyCount,
+ List<DataField> tableKeyFields,
+ List<DataField> tableValueFields,
+ int[] dataProjection,
+ int dataKeyCount,
+ List<DataField> dataKeyFields,
+ List<DataField> dataValueFields) {
+ List<Integer> tableKeyFieldIdList =
+ tableKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataKeyFieldIdList =
+ dataKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ int[] indexMapping = new int[tableProjection.length];
+
+ int[] dataKeyProjection = Arrays.copyOf(dataProjection, dataKeyCount);
+ for (int i = 0; i < tableKeyCount; i++) {
+ int fieldId = tableKeyFieldIdList.get(tableProjection[i]);
+ int dataFieldIndex = dataKeyFieldIdList.indexOf(fieldId);
+ indexMapping[i] = Ints.indexOf(dataKeyProjection, dataFieldIndex);
+ }
+ if (tableProjection.length >= tableKeyCount + 2) {
+ // seq and value kind
+ for (int i = tableKeyCount; i < tableKeyCount + 2; i++) {
+ indexMapping[i] = i + dataKeyCount - tableKeyCount;
+ }
+
+ int[] dataValueProjection =
+ Arrays.copyOfRange(dataProjection, dataKeyCount + 2, dataProjection.length);
+ for (int i = 0; i < dataValueProjection.length; i++) {
+ dataValueProjection[i] = dataValueProjection[i] - dataKeyFields.size() - 2;
+ }
+ List<Integer> tableValueFieldIdList =
+ tableValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataValueFieldIdList =
+ dataValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ for (int i = tableKeyCount + 2; i < tableProjection.length; i++) {
+ int fieldId =
+ tableValueFieldIdList.get(tableProjection[i] - tableKeyFields.size() - 2);
+ int dataFieldIndex = dataValueFieldIdList.indexOf(fieldId);
+ int dataValueIndex = Ints.indexOf(dataValueProjection, dataFieldIndex);
+ indexMapping[i] =
+ dataValueIndex < 0 ? dataValueIndex : dataValueIndex + dataKeyCount + 2;
+ }
+ }
+
+ for (int i = 0; i < indexMapping.length; i++) {
+ if (indexMapping[i] != i) {
+ return indexMapping;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Create data projection from table projection.
+ *
+ * @param tableFields the fields of table
+ * @param dataFields the fields of underlying data
+ * @param tableProjection the projection of table
+ * @return the projection of data
+ */
+ public static int[][] createDataProjection(
+ List<DataField> tableFields, List<DataField> dataFields, int[][] tableProjection) {
+ List<Integer> tableFieldIdList =
+ tableFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataFieldIdList =
+ dataFields.stream().map(DataField::id).collect(Collectors.toList());
+ return Arrays.stream(tableProjection)
+ .map(p -> Arrays.copyOf(p, p.length))
+ .peek(
+ p -> {
+ int fieldId = tableFieldIdList.get(p[0]);
+ p[0] = dataFieldIdList.indexOf(fieldId);
+ })
+ .filter(p -> p[0] >= 0)
+ .toArray(int[][]::new);
+ }
+
+ /**
+ * Create predicate list from data fields.
+ *
+ * @param tableFields the table fields
+ * @param dataFields the underlying data fields
+ * @param filters the filters
+ * @return the data filters
+ */
+ public static List<Predicate> createDataFilters(
+ List<DataField> tableFields, List<DataField> dataFields, List<Predicate> filters) {
+ if (filters == null) {
+ return null;
+ }
+
+ List<Predicate> dataFilters = new ArrayList<>(filters.size());
+ for (Predicate predicate : filters) {
+ dataFilters.add(createDataPredicate(tableFields, dataFields, predicate));
+ }
+ return dataFilters;
+ }
+
+ @Nullable
+ private static Predicate createDataPredicate(
+ List<DataField> tableFields, List<DataField> dataFields, Predicate predicate) {
+ if (predicate instanceof CompoundPredicate) {
+ CompoundPredicate compoundPredicate = (CompoundPredicate) predicate;
+ List<Predicate> children = compoundPredicate.children();
+ List<Predicate> dataChildren = new ArrayList<>(children.size());
+ for (Predicate child : children) {
+ Predicate dataPredicate = createDataPredicate(tableFields, dataFields, child);
+ if (dataPredicate != null) {
+ dataChildren.add(dataPredicate);
+ }
+ }
+ return new CompoundPredicate(compoundPredicate.function(), dataChildren);
+ } else if (predicate instanceof LeafPredicate) {
+ LeafPredicate leafPredicate = (LeafPredicate) predicate;
+ List<DataField> predicateTableFields =
+ tableFields.stream()
+ .filter(f -> f.name().equals(leafPredicate.fieldName()))
+ .collect(Collectors.toList());
+ if (predicateTableFields.size() != 1) {
+ throw new IllegalArgumentException(
+ String.format("Find none or multiple fields %s", predicateTableFields));
+ }
+ DataField tableField = predicateTableFields.get(0);
+ List<DataField> predicateDataFields =
+ dataFields.stream()
+ .filter(f -> f.id() == tableField.id())
+ .collect(Collectors.toList());
+ if (predicateDataFields.isEmpty()) {
+ return null;
+ } else if (predicateDataFields.size() > 1) {
+ throw new IllegalArgumentException(
+ String.format("Find none or multiple fields %s", predicateTableFields));
+ }
+ DataField dataField = predicateDataFields.get(0);
+ return new LeafPredicate(
+ leafPredicate.function(),
+ leafPredicate.type(),
+ dataFields.indexOf(dataField),
+ dataField.name(),
+ leafPredicate.literals());
Review Comment:
We can add TODO here.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] zjureel commented on pull request #396: [FLINK-27846] Schema evolution for reading data file
Posted by GitBox <gi...@apache.org>.
zjureel commented on PR #396:
URL: https://github.com/apache/flink-table-store/pull/396#issuecomment-1327161284
Hi @JingsongLi I have updated the codes and please help to review again, THX
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #396: [FLINK-27846] Schema evolution for reading data file
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #396:
URL: https://github.com/apache/flink-table-store/pull/396#discussion_r1033295182
##########
flink-table-store-core/src/test/java/org/apache/flink/table/store/table/SchemaEvolutionTableTestBase.java:
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.flink.table.store.table;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.store.CoreOptions;
+import org.apache.flink.table.store.file.schema.AtomicDataType;
+import org.apache.flink.table.store.file.schema.DataField;
+import org.apache.flink.table.store.file.schema.SchemaChange;
+import org.apache.flink.table.store.file.schema.SchemaManager;
+import org.apache.flink.table.store.file.schema.TableSchema;
+import org.apache.flink.table.store.file.schema.UpdateSchema;
+import org.apache.flink.table.store.file.utils.TestAtomicRenameFileSystem;
+import org.apache.flink.table.store.file.utils.TraceableFileSystem;
+import org.apache.flink.table.store.table.sink.TableCommit;
+import org.apache.flink.table.store.table.sink.TableWrite;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+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.Optional;
+import java.util.UUID;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Base test class for schema evolution in {@link FileStoreTable}. */
+public abstract class SchemaEvolutionTableTestBase {
Review Comment:
Can we extract a `TableTestBase` for `SchemaEvolutionTableTestBase` and `FileStoreTableTestBase`?
##########
flink-table-store-common/src/main/java/org/apache/flink/table/store/file/predicate/AlwaysFalse.java:
##########
@@ -0,0 +1,46 @@
+/*
+ * 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.flink.table.store.file.predicate;
+
+import org.apache.flink.table.store.format.FieldStats;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.Optional;
+
+/** Return false for all values. */
+public class AlwaysFalse extends LeafUnaryFunction {
+ public static final AlwaysFalse INSTANCE = new AlwaysFalse();
+
+ private AlwaysFalse() {}
+
+ @Override
+ public Optional<LeafFunction> negate() {
+ return Optional.of(AlwaysTrue.INSTANCE);
+ }
+
+ @Override
+ public boolean test(LogicalType type, Object value) {
Review Comment:
We can have better way to `AlwaysFalse` and `AlwaysTrue`, because they don't need to have field reference. But this requires more refactor.
Can you add TODO in the code and create a JIRA for this?
##########
flink-table-store-core/src/test/java/org/apache/flink/table/store/table/SchemaEvolutionTableTestBase.java:
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.flink.table.store.table;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.store.CoreOptions;
+import org.apache.flink.table.store.file.schema.AtomicDataType;
+import org.apache.flink.table.store.file.schema.DataField;
+import org.apache.flink.table.store.file.schema.SchemaChange;
+import org.apache.flink.table.store.file.schema.SchemaManager;
+import org.apache.flink.table.store.file.schema.TableSchema;
+import org.apache.flink.table.store.file.schema.UpdateSchema;
+import org.apache.flink.table.store.file.utils.TestAtomicRenameFileSystem;
+import org.apache.flink.table.store.file.utils.TraceableFileSystem;
+import org.apache.flink.table.store.table.sink.TableCommit;
+import org.apache.flink.table.store.table.sink.TableWrite;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+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.Optional;
+import java.util.UUID;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Base test class for schema evolution in {@link FileStoreTable}. */
+public abstract class SchemaEvolutionTableTestBase {
+ protected static final List<DataField> SCHEMA_0_FIELDS =
+ Arrays.asList(
+ new DataField(0, "a", new AtomicDataType(DataTypes.STRING().getLogicalType())),
+ new DataField(1, "pt", new AtomicDataType(DataTypes.INT().getLogicalType())),
+ new DataField(2, "b", new AtomicDataType(DataTypes.INT().getLogicalType())),
+ new DataField(3, "c", new AtomicDataType(DataTypes.STRING().getLogicalType())),
+ new DataField(4, "kt", new AtomicDataType(DataTypes.BIGINT().getLogicalType())),
+ new DataField(5, "d", new AtomicDataType(DataTypes.STRING().getLogicalType())));
+ protected static final List<DataField> SCHEMA_1_FIELDS =
+ Arrays.asList(
+ new DataField(1, "pt", new AtomicDataType(DataTypes.INT().getLogicalType())),
+ new DataField(2, "d", new AtomicDataType(DataTypes.INT().getLogicalType())),
+ new DataField(4, "kt", new AtomicDataType(DataTypes.BIGINT().getLogicalType())),
+ new DataField(6, "a", new AtomicDataType(DataTypes.INT().getLogicalType())),
+ new DataField(7, "f", new AtomicDataType(DataTypes.STRING().getLogicalType())),
+ new DataField(8, "b", new AtomicDataType(DataTypes.STRING().getLogicalType())));
+ protected static final List<String> PARTITION_NAMES = Collections.singletonList("pt");
+ protected static final List<String> PRIMARY_KEY_NAMES = Arrays.asList("pt", "kt");
+
+ protected Path tablePath;
+ protected String commitUser;
+ protected final Configuration tableConfig = new Configuration();
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @BeforeEach
+ public void before() throws Exception {
+ tablePath = new Path(TestAtomicRenameFileSystem.SCHEME + "://" + tempDir.toString());
+ commitUser = UUID.randomUUID().toString();
+ tableConfig.set(CoreOptions.PATH, tablePath.toString());
+ tableConfig.set(CoreOptions.BUCKET, 2);
+ }
+
+ @AfterEach
+ public void after() throws IOException {
+ // assert all connections are closed
+ FileSystem fileSystem = tablePath.getFileSystem();
+ assertThat(fileSystem).isInstanceOf(TraceableFileSystem.class);
+ TraceableFileSystem traceableFileSystem = (TraceableFileSystem) fileSystem;
+
+ java.util.function.Predicate<Path> pathPredicate =
+ path -> path.toString().contains(tempDir.toString());
+ assertThat(traceableFileSystem.openInputStreams(pathPredicate)).isEmpty();
+ assertThat(traceableFileSystem.openOutputStreams(pathPredicate)).isEmpty();
+ }
+
+ protected List<String> getPrimaryKeyNames() {
+ return PRIMARY_KEY_NAMES;
+ }
+
+ protected abstract FileStoreTable createFileStoreTable(Map<Long, TableSchema> tableSchemas);
+
+ public static <R> void writeAndCheckFileResult(
+ Function<Map<Long, TableSchema>, R> firstChecker,
+ BiConsumer<R, Map<Long, TableSchema>> secondChecker,
+ List<String> primaryKeyNames,
+ Configuration tableConfig,
+ Function<Map<Long, TableSchema>, FileStoreTable> createFileStoreTable)
+ throws Exception {
+ Map<Long, TableSchema> tableSchemas = new HashMap<>();
+ tableSchemas.put(
+ 0L,
+ new TableSchema(
+ 0,
+ SCHEMA_0_FIELDS,
+ 5,
+ PARTITION_NAMES,
+ primaryKeyNames,
+ tableConfig.toMap(),
+ ""));
+ FileStoreTable table = createFileStoreTable.apply(tableSchemas);
+ TableWrite write = table.newWrite("user");
+ TableCommit commit = table.newCommit("user");
+
+ write.write(
+ GenericRowData.of(
+ StringData.fromString("S001"),
Review Comment:
Maybe we can create a util method, `GenericRowData rowData(Object... values)`, we iterate over each element. If there is a String, it will be converted to StringData.
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaEvolutionUtil.java:
##########
@@ -62,4 +88,240 @@ public static int[] createIndexMapping(
}
return null;
}
+
+ /**
+ * Create index mapping from table projection to underlying data projection. For example, the
+ * table and data fields are as follows
+ *
+ * <ul>
+ * <li>table fields: 1->c, 3->a, 4->e, 5->d, 6->b
+ * <li>data fields: 1->a, 2->b, 3->c, 4->d
+ * </ul>
+ *
+ * <p>The table and data top projections are as follows
+ *
+ * <ul>
+ * <li>table projection: [0, 4, 1]
+ * <li>data projection: [0, 2]
+ * </ul>
+ *
+ * <p>We can first get fields list for table and data projections from their fields as follows
+ *
+ * <ul>
+ * <li>table projection field list: [1->c, 6->b, 3->a]
+ * <li>data projection field list: [1->a, 3->c]
+ * </ul>
+ *
+ * <p>Then create index mapping based on the fields list.
+ *
+ * <p>/// TODO should support nest index mapping when nest schema evolution is supported.
+ *
+ * @param tableProjection the table projection
+ * @param tableFields the fields in table
+ * @param dataProjection the underlying data projection
+ * @param dataFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ List<DataField> tableFields,
+ int[] dataProjection,
+ List<DataField> dataFields) {
+ List<DataField> tableProjectFields = new ArrayList<>(tableProjection.length);
+ for (int index : tableProjection) {
+ tableProjectFields.add(tableFields.get(index));
+ }
+
+ List<DataField> dataProjectFields = new ArrayList<>(dataProjection.length);
+ for (int index : dataProjection) {
+ dataProjectFields.add(dataFields.get(index));
+ }
+
+ return createIndexMapping(tableProjectFields, dataProjectFields);
+ }
+
+ /**
+ * Create index mapping from table projection to data with key and value fields. We should first
+ * create table and data fields with their key/value fields, then create index mapping with
+ * their projections and fields. For example, the table and data projections and fields are as
+ * follows
+ *
+ * <ul>
+ * <li>Table key fields: 1->ka, 3->kb, 5->kc, 6->kd; value fields: 0->a, 2->d, 4->b;
+ * projection: [0, 2, 3, 4, 5, 7] where 0 is 1->ka, 2 is 5->kc, 3 is 5->kc, 4/5 are seq
+ * and kind, 7 is 2->d
+ * <li>Data key fields: 1->kb, 5->ka; value fields: 2->aa, 4->f; projection: [0, 1, 2, 3, 4]
+ * where 0 is 1->kb, 1 is 5->ka, 2/3 are seq and kind, 4 is 2->aa
+ * </ul>
+ *
+ * <p>First we will get max key id from table and data fields which is 6, then create table and
+ * data fields on it
+ *
+ * <ul>
+ * <li>Table fields: 1->ka, 3->kb, 5->kc, 6->kd, 7->seq, 8->kind, 9->a, 11->d, 13->b
+ * <li>Data fields: 1->kb, 5->ka, 7->seq, 8->kind, 11->aa, 13->f
+ * </ul>
+ *
+ * <p>Finally we can create index mapping with table/data projections and fields.
+ *
+ * <p>/// TODO should support nest index mapping when nest schema evolution is supported.
+ *
+ * @param tableProjection the table projection
+ * @param tableKeyFields the table key fields
+ * @param tableValueFields the table value fields
+ * @param dataProjection the data projection
+ * @param dataKeyFields the data key fields
+ * @param dataValueFields the data value fields
+ * @return the result index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ List<DataField> tableKeyFields,
+ List<DataField> tableValueFields,
+ int[] dataProjection,
+ List<DataField> dataKeyFields,
+ List<DataField> dataValueFields) {
+ int maxKeyId =
+ Math.max(
+ tableKeyFields.stream().mapToInt(DataField::id).max().orElse(0),
+ dataKeyFields.stream().mapToInt(DataField::id).max().orElse(0));
+ List<DataField> tableFields =
+ KeyValue.createKeyValueFields(tableKeyFields, tableValueFields, maxKeyId);
+ List<DataField> dataFields =
+ KeyValue.createKeyValueFields(dataKeyFields, dataValueFields, maxKeyId);
+ return createIndexMapping(tableProjection, tableFields, dataProjection, dataFields);
+ }
+
+ /**
+ * Create data projection from table projection. For example, the table and data fields are as
+ * follows
+ *
+ * <ul>
+ * <li>table fields: 1->c, 3->a, 4->e, 5->d, 6->b
+ * <li>data fields: 1->a, 2->b, 3->c, 4->d
+ * </ul>
+ *
+ * <p>When we project 1->c, 6->b, 3->a from table fields, the table projection is [[0], [4],
+ * [1]], in which 0 is the index of field 1->c, 4 is the index of field 6->b, 1 is the index of
+ * field 3->a in table fields. We need to create data projection from [[0], [4], [1]] as
+ * follows:
+ *
+ * <ul>
+ * <li>Get field id of each index in table projection from table fields
+ * <li>Get index of each field above from data fields
+ * </ul>
+ *
+ * <p>The we can create table projection as follows: [[0], [-1], [2]], in which 0, -1 and 2 are
+ * the index of fields [1->c, 6->b, 3->a] in data fields. When we project column from underlying
+ * data, we need to specify the field index and name. It is difficult to assign a proper field
+ * id and name for 6->b in data projection and add it to data fields, and we can't use 6->b
+ * directly because the field index of b in underlying is 2. We can remove the -1 field index in
+ * data projection, then the result data projection is: [[0], [2]].
+ *
+ * <p>We create {@link RowData} for 1->a, 3->c after projecting them from underlying data, then
+ * create {@link ProjectedRowData} with a index mapping and return null for 6->b in table
+ * fields.
+ *
+ * @param tableFields the fields of table
+ * @param dataFields the fields of underlying data
+ * @param tableProjection the projection of table
+ * @return the projection of data
+ */
+ public static int[][] createDataProjection(
+ List<DataField> tableFields, List<DataField> dataFields, int[][] tableProjection) {
+ List<Integer> dataFieldIdList =
+ dataFields.stream().map(DataField::id).collect(Collectors.toList());
+ return Arrays.stream(tableProjection)
+ .map(p -> Arrays.copyOf(p, p.length))
+ .peek(
+ p -> {
+ int fieldId = tableFields.get(p[0]).id();
+ p[0] = dataFieldIdList.indexOf(fieldId);
+ })
+ .filter(p -> p[0] >= 0)
+ .toArray(int[][]::new);
+ }
+
+ /**
+ * Create predicate list from data fields. We will visit all predicate in filters, reset it's
+ * field index, name and type, and use {@link AlwaysFalse} or {@link AlwaysTrue} if the field is
+ * not exist.
+ *
+ * @param tableFields the table fields
+ * @param dataFields the underlying data fields
+ * @param filters the filters
+ * @return the data filters
+ */
+ @Nullable
+ public static List<Predicate> createDataFilters(
+ List<DataField> tableFields, List<DataField> dataFields, List<Predicate> filters) {
Review Comment:
Maybe we can construct them first like:
- `Map<String, DataField> tableFields`?
- `LinkedHashMap<Integer, DataField> dataFields` and we can create a `indexOf` for `LinkedHashMap`?
##########
flink-table-store-core/src/test/java/org/apache/flink/table/store/table/ChangelogWithKeyFileMetaFilterTest.java:
##########
@@ -46,7 +46,7 @@ public void before() throws Exception {
@Test
@Override
public void testTableScan() throws Exception {
- writeAndCheckFileMeta(
+ writeAndCheckFileResult(
Review Comment:
Why need to override three methods? I think we can add comments to explain in the code.
##########
flink-table-store-core/src/test/java/org/apache/flink/table/store/table/FileDataFilterTestBase.java:
##########
@@ -0,0 +1,460 @@
+/*
+ * 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.flink.table.store.table;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.store.file.mergetree.compact.ConcatRecordReader;
+import org.apache.flink.table.store.file.predicate.Equal;
+import org.apache.flink.table.store.file.predicate.IsNull;
+import org.apache.flink.table.store.file.predicate.LeafPredicate;
+import org.apache.flink.table.store.file.predicate.Predicate;
+import org.apache.flink.table.store.file.predicate.PredicateBuilder;
+import org.apache.flink.table.store.file.schema.RowDataType;
+import org.apache.flink.table.store.file.utils.RecordReader;
+import org.apache.flink.table.store.file.utils.RecordReaderIterator;
+import org.apache.flink.table.store.table.source.Split;
+import org.apache.flink.table.store.table.source.TableRead;
+import org.apache.flink.types.RowKind;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Base test class of file data for schema evolution in {@link FileStoreTable}. */
+public abstract class FileDataFilterTestBase extends SchemaEvolutionTableTestBase {
+
+ protected static final int[] PROJECTION = new int[] {3, 2, 1};
+
+ protected static final Function<RowData, String> SCHEMA_0_ROW_TO_STRING =
+ rowData ->
+ getNullOrString(rowData, 0)
+ + "|"
+ + getNullOrInt(rowData, 1)
+ + "|"
+ + getNullOrInt(rowData, 2)
+ + "|"
+ + getNullOrString(rowData, 3)
+ + "|"
+ + getNullOrLong(rowData, 4)
+ + "|"
+ + getNullOrString(rowData, 5);
+
+ protected static final Function<RowData, String> STREAMING_SCHEMA_0_ROW_TO_STRING =
+ rowData ->
+ (rowData.getRowKind() == RowKind.INSERT ? "+" : "-")
+ + getNullOrString(rowData, 0)
+ + "|"
+ + getNullOrInt(rowData, 1)
+ + "|"
+ + getNullOrInt(rowData, 2)
+ + "|"
+ + getNullOrString(rowData, 3)
+ + "|"
+ + getNullOrLong(rowData, 4)
+ + "|"
+ + getNullOrString(rowData, 5);
+
+ protected static final Function<RowData, String> SCHEMA_0_PROJECT_ROW_TO_STRING =
+ rowData ->
+ getNullOrString(rowData, 0)
+ + "|"
+ + getNullOrInt(rowData, 1)
+ + "|"
+ + getNullOrInt(rowData, 2);
+
+ protected static final Function<RowData, String> STREAMING_SCHEMA_0_PROJECT_ROW_TO_STRING =
+ rowData ->
+ (rowData.getRowKind() == RowKind.INSERT ? "+" : "-")
+ + getNullOrString(rowData, 0)
+ + "|"
+ + getNullOrInt(rowData, 1)
+ + "|"
+ + getNullOrInt(rowData, 2);
+
+ protected static final Function<RowData, String> SCHEMA_1_ROW_TO_STRING =
+ rowData ->
+ getNullOrInt(rowData, 0)
+ + "|"
+ + getNullOrInt(rowData, 1)
+ + "|"
+ + getNullOrLong(rowData, 2)
+ + "|"
+ + getNullOrInt(rowData, 3)
+ + "|"
+ + getNullOrString(rowData, 4)
+ + "|"
+ + getNullOrString(rowData, 5);
+
+ protected static final Function<RowData, String> STREAMING_SCHEMA_1_ROW_TO_STRING =
+ rowData ->
+ (rowData.getRowKind() == RowKind.INSERT ? "+" : "-")
+ + getNullOrInt(rowData, 0)
+ + "|"
+ + getNullOrInt(rowData, 1)
+ + "|"
+ + getNullOrLong(rowData, 2)
+ + "|"
+ + getNullOrInt(rowData, 3)
+ + "|"
+ + getNullOrString(rowData, 4)
+ + "|"
+ + getNullOrString(rowData, 5);
+
+ protected static final Function<RowData, String> SCHEMA_1_PROJECT_ROW_TO_STRING =
+ rowData ->
+ getNullOrInt(rowData, 0)
+ + "|"
+ + getNullOrLong(rowData, 1)
+ + "|"
+ + getNullOrInt(rowData, 2);
+
+ protected static final Function<RowData, String> STREAMING_SCHEMA_1_PROJECT_ROW_TO_STRING =
+ rowData ->
+ (rowData.getRowKind() == RowKind.INSERT ? "+" : "-")
+ + getNullOrInt(rowData, 0)
+ + "|"
+ + getNullOrLong(rowData, 1)
+ + "|"
+ + getNullOrInt(rowData, 2);
+
+ private static String getNullOrInt(RowData rowData, int index) {
+ return rowData.isNullAt(index) ? "null" : String.valueOf(rowData.getInt(index));
+ }
+
+ private static String getNullOrLong(RowData rowData, int index) {
+ return rowData.isNullAt(index) ? "null" : String.valueOf(rowData.getLong(index));
+ }
+
+ private static String getNullOrString(RowData rowData, int index) {
+ return rowData.isNullAt(index) ? "null" : rowData.getString(index).toString();
+ }
+
+ @Test
+ public void testReadFilterExistField() throws Exception {
+ writeAndCheckFileResult(
+ schemas -> {
+ PredicateBuilder builder =
+ new PredicateBuilder(RowDataType.toRowType(false, SCHEMA_0_FIELDS));
+ FileStoreTable table = createFileStoreTable(schemas);
+ List<Split> splits = table.newScan().plan().splits();
+ // filter with "b" = 15 in schema0
+ TableRead read = table.newRead().withFilter(builder.equal(2, 15));
+
+ assertThat(getResult(read, splits, SCHEMA_0_ROW_TO_STRING))
+ .hasSameElementsAs(
+ Arrays.asList(
+ "S005|2|15|S15|115|S115", "S006|2|16|S16|116|S116"));
+ return null;
+ },
+ (files, schemas) -> {
+ PredicateBuilder builder =
+ new PredicateBuilder(RowDataType.toRowType(false, SCHEMA_1_FIELDS));
+ FileStoreTable table = createFileStoreTable(schemas);
+ List<Split> splits = table.newScan().plan().splits();
+
+ // filter with "d" = 15 in schema1 which should be mapped to "b" = 15 in schema0
+ TableRead read1 = table.newRead().withFilter(builder.equal(1, 15));
+ assertThat(getResult(read1, splits, SCHEMA_1_ROW_TO_STRING))
+ .hasSameElementsAs(
+ Arrays.asList(
+ "2|15|115|null|null|null", "2|16|116|null|null|null"));
+
+ // filter with "d" = 21 in schema1
+ TableRead read2 = table.newRead().withFilter(builder.equal(1, 21));
+ assertThat(getResult(read2, splits, SCHEMA_1_ROW_TO_STRING))
+ .hasSameElementsAs(
+ Arrays.asList(
+ "1|21|121|1121|S011|S21", "1|22|122|1122|S012|S22"));
+ },
+ getPrimaryKeyNames(),
+ tableConfig,
+ this::createFileStoreTable);
+ }
+
+ @Test
+ public void testReadFilterNonExistField() throws Exception {
+ writeAndCheckFileResult(
+ schemas -> null,
+ (files, schemas) -> {
+ PredicateBuilder builder =
+ new PredicateBuilder(RowDataType.toRowType(false, SCHEMA_1_FIELDS));
+ FileStoreTable table = createFileStoreTable(schemas);
+ List<Split> splits = table.newScan().plan().splits();
+
+ // filter with "a" = 1122 in schema1 which is not exist in schema0
+ TableRead read1 = table.newRead().withFilter(builder.equal(3, 1122));
+ assertThat(getResult(read1, splits, SCHEMA_1_ROW_TO_STRING))
+ .hasSameElementsAs(
+ Arrays.asList(
+ "2|12|112|null|null|null",
+ "2|15|115|null|null|null",
+ "2|16|116|null|null|null",
+ "1|11|111|null|null|null",
+ "1|13|113|null|null|null",
+ "1|14|114|null|null|null",
+ "1|21|121|1121|S011|S21",
+ "1|22|122|1122|S012|S22"));
+
+ // filter with "a" = 1122 in scan and read
+ splits = table.newScan().withFilter(builder.equal(3, 1122)).plan().splits();
+ TableRead read2 = table.newRead().withFilter(builder.equal(3, 1122));
+ assertThat(getResult(read2, splits, SCHEMA_1_ROW_TO_STRING))
+ .hasSameElementsAs(
+ Arrays.asList(
+ "1|21|121|1121|S011|S21", "1|22|122|1122|S012|S22"));
+ },
+ getPrimaryKeyNames(),
+ tableConfig,
+ this::createFileStoreTable);
+ }
+
+ @Test
+ public void testReadFilterMultipleFields() throws Exception {
+ writeAndCheckFileResult(
+ schemas -> null,
+ (files, schemas) -> {
+ List<Predicate> predicateList =
+ Arrays.asList(
+ new LeafPredicate(
+ Equal.INSTANCE,
+ DataTypes.INT().getLogicalType(),
+ 1,
+ "d",
+ Arrays.asList(21)),
Review Comment:
`Collections.singletonList`
##########
flink-table-store-core/src/test/java/org/apache/flink/table/store/table/SchemaEvolutionTableTestBase.java:
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.flink.table.store.table;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.store.CoreOptions;
+import org.apache.flink.table.store.file.schema.AtomicDataType;
+import org.apache.flink.table.store.file.schema.DataField;
+import org.apache.flink.table.store.file.schema.SchemaChange;
+import org.apache.flink.table.store.file.schema.SchemaManager;
+import org.apache.flink.table.store.file.schema.TableSchema;
+import org.apache.flink.table.store.file.schema.UpdateSchema;
+import org.apache.flink.table.store.file.utils.TestAtomicRenameFileSystem;
+import org.apache.flink.table.store.file.utils.TraceableFileSystem;
+import org.apache.flink.table.store.table.sink.TableCommit;
+import org.apache.flink.table.store.table.sink.TableWrite;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+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.Optional;
+import java.util.UUID;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Base test class for schema evolution in {@link FileStoreTable}. */
+public abstract class SchemaEvolutionTableTestBase {
+ protected static final List<DataField> SCHEMA_0_FIELDS =
+ Arrays.asList(
+ new DataField(0, "a", new AtomicDataType(DataTypes.STRING().getLogicalType())),
+ new DataField(1, "pt", new AtomicDataType(DataTypes.INT().getLogicalType())),
+ new DataField(2, "b", new AtomicDataType(DataTypes.INT().getLogicalType())),
+ new DataField(3, "c", new AtomicDataType(DataTypes.STRING().getLogicalType())),
+ new DataField(4, "kt", new AtomicDataType(DataTypes.BIGINT().getLogicalType())),
+ new DataField(5, "d", new AtomicDataType(DataTypes.STRING().getLogicalType())));
+ protected static final List<DataField> SCHEMA_1_FIELDS =
+ Arrays.asList(
+ new DataField(1, "pt", new AtomicDataType(DataTypes.INT().getLogicalType())),
+ new DataField(2, "d", new AtomicDataType(DataTypes.INT().getLogicalType())),
+ new DataField(4, "kt", new AtomicDataType(DataTypes.BIGINT().getLogicalType())),
+ new DataField(6, "a", new AtomicDataType(DataTypes.INT().getLogicalType())),
+ new DataField(7, "f", new AtomicDataType(DataTypes.STRING().getLogicalType())),
+ new DataField(8, "b", new AtomicDataType(DataTypes.STRING().getLogicalType())));
+ protected static final List<String> PARTITION_NAMES = Collections.singletonList("pt");
+ protected static final List<String> PRIMARY_KEY_NAMES = Arrays.asList("pt", "kt");
+
+ protected Path tablePath;
+ protected String commitUser;
+ protected final Configuration tableConfig = new Configuration();
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @BeforeEach
+ public void before() throws Exception {
+ tablePath = new Path(TestAtomicRenameFileSystem.SCHEME + "://" + tempDir.toString());
+ commitUser = UUID.randomUUID().toString();
+ tableConfig.set(CoreOptions.PATH, tablePath.toString());
+ tableConfig.set(CoreOptions.BUCKET, 2);
+ }
+
+ @AfterEach
+ public void after() throws IOException {
+ // assert all connections are closed
+ FileSystem fileSystem = tablePath.getFileSystem();
+ assertThat(fileSystem).isInstanceOf(TraceableFileSystem.class);
+ TraceableFileSystem traceableFileSystem = (TraceableFileSystem) fileSystem;
+
+ java.util.function.Predicate<Path> pathPredicate =
+ path -> path.toString().contains(tempDir.toString());
+ assertThat(traceableFileSystem.openInputStreams(pathPredicate)).isEmpty();
+ assertThat(traceableFileSystem.openOutputStreams(pathPredicate)).isEmpty();
+ }
+
+ protected List<String> getPrimaryKeyNames() {
+ return PRIMARY_KEY_NAMES;
+ }
+
+ protected abstract FileStoreTable createFileStoreTable(Map<Long, TableSchema> tableSchemas);
+
+ public static <R> void writeAndCheckFileResult(
+ Function<Map<Long, TableSchema>, R> firstChecker,
+ BiConsumer<R, Map<Long, TableSchema>> secondChecker,
+ List<String> primaryKeyNames,
+ Configuration tableConfig,
+ Function<Map<Long, TableSchema>, FileStoreTable> createFileStoreTable)
+ throws Exception {
+ Map<Long, TableSchema> tableSchemas = new HashMap<>();
+ tableSchemas.put(
+ 0L,
+ new TableSchema(
+ 0,
+ SCHEMA_0_FIELDS,
+ 5,
+ PARTITION_NAMES,
+ primaryKeyNames,
+ tableConfig.toMap(),
+ ""));
+ FileStoreTable table = createFileStoreTable.apply(tableSchemas);
+ TableWrite write = table.newWrite("user");
+ TableCommit commit = table.newCommit("user");
+
+ write.write(
+ GenericRowData.of(
+ StringData.fromString("S001"),
+ 1,
+ 11,
+ StringData.fromString("S11"),
+ 111L,
+ StringData.fromString("S111")));
+ write.write(
+ GenericRowData.of(
+ StringData.fromString("S002"),
+ 2,
+ 12,
+ StringData.fromString("S12"),
+ 112L,
+ StringData.fromString("S112")));
+ write.write(
+ GenericRowData.of(
+ StringData.fromString("S003"),
+ 1,
+ 13,
+ StringData.fromString("S13"),
+ 113L,
+ StringData.fromString("S113")));
+ commit.commit(0, write.prepareCommit(true, 0));
+
+ write.write(
+ GenericRowData.of(
+ StringData.fromString("S004"),
+ 1,
+ 14,
+ StringData.fromString("S14"),
+ 114L,
+ StringData.fromString("S114")));
+ write.write(
+ GenericRowData.of(
+ StringData.fromString("S005"),
+ 2,
+ 15,
+ StringData.fromString("S15"),
+ 115L,
+ StringData.fromString("S115")));
+ write.write(
+ GenericRowData.of(
+ StringData.fromString("S006"),
+ 2,
+ 16,
+ StringData.fromString("S16"),
+ 116L,
+ StringData.fromString("S116")));
+ commit.commit(0, write.prepareCommit(true, 0));
+ write.close();
+ R result = firstChecker.apply(tableSchemas);
+
+ tableSchemas.put(
Review Comment:
Add comments to explain fields removing and fields adding, this can help to understand the test 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #396: [FLINK-27846] Schema evolution for reading data file
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #396:
URL: https://github.com/apache/flink-table-store/pull/396#discussion_r1031291563
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/BulkFormatMapping.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * 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.flink.table.store.file.utils;
+
+import org.apache.flink.connector.file.src.FileSourceSplit;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.store.file.KeyValue;
+import org.apache.flink.table.store.file.predicate.Predicate;
+import org.apache.flink.table.store.file.schema.DataField;
+import org.apache.flink.table.store.file.schema.KeyValueFieldsExtractor;
+import org.apache.flink.table.store.file.schema.RowDataType;
+import org.apache.flink.table.store.file.schema.SchemaEvolutionUtil;
+import org.apache.flink.table.store.file.schema.TableSchema;
+import org.apache.flink.table.store.format.FileFormat;
+import org.apache.flink.table.store.utils.Projection;
+import org.apache.flink.table.types.logical.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/** Class with index mapping and bulk format. */
+public class BulkFormatMapping {
+ @Nullable private final int[] indexMapping;
+ private final BulkFormat<RowData, FileSourceSplit> bulkFormat;
+
+ public BulkFormatMapping(int[] indexMapping, BulkFormat<RowData, FileSourceSplit> bulkFormat) {
+ this.indexMapping = indexMapping;
+ this.bulkFormat = bulkFormat;
+ }
+
+ @Nullable
+ public int[] getIndexMapping() {
+ return indexMapping;
+ }
+
+ public BulkFormat<RowData, FileSourceSplit> getReaderFactory() {
+ return bulkFormat;
+ }
+
+ public static BulkFormatMappingBuilder newBuilder(
+ FileFormat fileFormat,
+ KeyValueFieldsExtractor extractor,
+ int[][] keyProjection,
+ int[][] valueProjection,
+ int[][] projection,
+ @Nullable List<Predicate> filters) {
+ return new BulkFormatMappingBuilder(
+ fileFormat, extractor, keyProjection, valueProjection, projection, filters);
+ }
+
+ /** Builder to build {@link BulkFormatMapping}. */
+ public static class BulkFormatMappingBuilder {
+ private final FileFormat fileFormat;
+ private final KeyValueFieldsExtractor extractor;
+ private final int[][] keyProjection;
+ private final int[][] valueProjection;
+ private final int[][] projection;
+ @Nullable private final List<Predicate> filters;
+
+ private BulkFormatMappingBuilder(
+ FileFormat fileFormat,
+ KeyValueFieldsExtractor extractor,
+ int[][] keyProjection,
+ int[][] valueProjection,
+ int[][] projection,
+ @Nullable List<Predicate> filters) {
+ this.fileFormat = fileFormat;
+ this.extractor = extractor;
+ this.keyProjection = keyProjection;
+ this.valueProjection = valueProjection;
+ this.projection = projection;
+ this.filters = filters;
+ }
+
+ public BulkFormatMapping build(TableSchema tableSchema, TableSchema dataSchema) {
+ List<DataField> tableKeyFields = extractor.keyFields(tableSchema);
+ List<DataField> tableValueFields = extractor.valueFields(tableSchema);
+
+ List<DataField> dataKeyFields = extractor.keyFields(dataSchema);
+ List<DataField> dataValueFields = extractor.valueFields(dataSchema);
+ int[][] dataKeyProjection =
+ SchemaEvolutionUtil.createDataProjection(
+ tableKeyFields, dataKeyFields, keyProjection);
+ int[][] dataValueProjection =
+ SchemaEvolutionUtil.createDataProjection(
+ tableValueFields, dataValueFields, valueProjection);
+
+ RowType keyType = RowDataType.toRowType(false, dataKeyFields);
+ RowType valueType = RowDataType.toRowType(false, dataValueFields);
+ RowType dataRecordType = KeyValue.schema(keyType, valueType);
+ int[][] dataProjection =
+ KeyValue.project(
+ dataKeyProjection, dataValueProjection, keyType.getFieldCount());
+
+ int[] indexMapping =
+ SchemaEvolutionUtil.createIndexMapping(
Review Comment:
This process is really complicated, and I'm crazy.
We can add more comments to explain every step.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] zjureel commented on a diff in pull request #396: [FLINK-27846] Schema evolution for reading data file
Posted by GitBox <gi...@apache.org>.
zjureel commented on code in PR #396:
URL: https://github.com/apache/flink-table-store/pull/396#discussion_r1032107447
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/schema/SchemaEvolutionUtil.java:
##########
@@ -62,4 +72,193 @@ public static int[] createIndexMapping(
}
return null;
}
+
+ /**
+ * Create index mapping from table projection to underlying data projection.
+ *
+ * @param tableProjection the table projection
+ * @param tableFields the fields in table
+ * @param dataProjection the underlying data projection
+ * @param dataFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ List<DataField> tableFields,
+ int[] dataProjection,
+ List<DataField> dataFields) {
+ return createIndexMapping(
+ tableProjection,
+ tableProjection.length,
+ tableFields,
+ Collections.emptyList(),
+ dataProjection,
+ dataProjection.length,
+ dataFields,
+ Collections.emptyList());
+ }
+
+ /**
+ * Create index mapping from table projection to underlying data projection for key value.
+ *
+ * @param tableProjection the table projection
+ * @param tableKeyCount the key count in table
+ * @param tableKeyFields the key fields in table
+ * @param tableValueFields the value fields in table
+ * @param dataProjection the data projection
+ * @param dataKeyCount the data key count
+ * @param dataKeyFields the fields in underlying data
+ * @param dataValueFields the fields in underlying data
+ * @return the index mapping
+ */
+ @Nullable
+ public static int[] createIndexMapping(
+ int[] tableProjection,
+ int tableKeyCount,
+ List<DataField> tableKeyFields,
+ List<DataField> tableValueFields,
+ int[] dataProjection,
+ int dataKeyCount,
+ List<DataField> dataKeyFields,
+ List<DataField> dataValueFields) {
+ List<Integer> tableKeyFieldIdList =
+ tableKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataKeyFieldIdList =
+ dataKeyFields.stream().map(DataField::id).collect(Collectors.toList());
+ int[] indexMapping = new int[tableProjection.length];
+
+ int[] dataKeyProjection = Arrays.copyOf(dataProjection, dataKeyCount);
+ for (int i = 0; i < tableKeyCount; i++) {
+ int fieldId = tableKeyFieldIdList.get(tableProjection[i]);
+ int dataFieldIndex = dataKeyFieldIdList.indexOf(fieldId);
+ indexMapping[i] = Ints.indexOf(dataKeyProjection, dataFieldIndex);
+ }
+ if (tableProjection.length >= tableKeyCount + 2) {
+ // seq and value kind
+ for (int i = tableKeyCount; i < tableKeyCount + 2; i++) {
+ indexMapping[i] = i + dataKeyCount - tableKeyCount;
+ }
+
+ int[] dataValueProjection =
+ Arrays.copyOfRange(dataProjection, dataKeyCount + 2, dataProjection.length);
+ for (int i = 0; i < dataValueProjection.length; i++) {
+ dataValueProjection[i] = dataValueProjection[i] - dataKeyFields.size() - 2;
+ }
+ List<Integer> tableValueFieldIdList =
+ tableValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ List<Integer> dataValueFieldIdList =
+ dataValueFields.stream().map(DataField::id).collect(Collectors.toList());
+ for (int i = tableKeyCount + 2; i < tableProjection.length; i++) {
+ int fieldId =
+ tableValueFieldIdList.get(tableProjection[i] - tableKeyFields.size() - 2);
+ int dataFieldIndex = dataValueFieldIdList.indexOf(fieldId);
+ int dataValueIndex = Ints.indexOf(dataValueProjection, dataFieldIndex);
+ indexMapping[i] =
+ dataValueIndex < 0 ? dataValueIndex : dataValueIndex + dataKeyCount + 2;
+ }
+ }
+
+ for (int i = 0; i < indexMapping.length; i++) {
+ if (indexMapping[i] != i) {
+ return indexMapping;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Create data projection from table projection.
+ *
+ * @param tableFields the fields of table
+ * @param dataFields the fields of underlying data
+ * @param tableProjection the projection of table
+ * @return the projection of data
+ */
+ public static int[][] createDataProjection(
+ List<DataField> tableFields, List<DataField> dataFields, int[][] tableProjection) {
+ List<Integer> tableFieldIdList =
Review Comment:
@JingsongLi Table Store doesn't support nest projection now? If it does, maybe we don't need to check nest projection here, WDYT?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on pull request #396: [FLINK-27846] Schema evolution for reading data file
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on PR #396:
URL: https://github.com/apache/flink-table-store/pull/396#issuecomment-1323264964
Thanks @zjureel , very good job! I will take a look~
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] zjureel commented on a diff in pull request #396: [FLINK-27846] Schema evolution for reading data file
Posted by GitBox <gi...@apache.org>.
zjureel commented on code in PR #396:
URL: https://github.com/apache/flink-table-store/pull/396#discussion_r1033422273
##########
flink-table-store-common/src/main/java/org/apache/flink/table/store/file/predicate/AlwaysFalse.java:
##########
@@ -0,0 +1,46 @@
+/*
+ * 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.flink.table.store.file.predicate;
+
+import org.apache.flink.table.store.format.FieldStats;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.Optional;
+
+/** Return false for all values. */
+public class AlwaysFalse extends LeafUnaryFunction {
+ public static final AlwaysFalse INSTANCE = new AlwaysFalse();
+
+ private AlwaysFalse() {}
+
+ @Override
+ public Optional<LeafFunction> negate() {
+ return Optional.of(AlwaysTrue.INSTANCE);
+ }
+
+ @Override
+ public boolean test(LogicalType type, Object value) {
Review Comment:
Added https://issues.apache.org/jira/browse/FLINK-30227
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] zjureel commented on pull request #396: [FLINK-27846] Schema evolution for reading data file
Posted by GitBox <gi...@apache.org>.
zjureel commented on PR #396:
URL: https://github.com/apache/flink-table-store/pull/396#issuecomment-1328922731
Thanks @JingsongLi I have updated the codes
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org