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/24 07:55:46 UTC
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #396: [FLINK-27846] Schema evolution for reading data file
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