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/12/12 04:48:04 UTC
[GitHub] [flink-table-store] tsreaper commented on a diff in pull request #428: [FLINK-30341] Introduce audit_log system table
tsreaper commented on code in PR #428:
URL: https://github.com/apache/flink-table-store/pull/428#discussion_r1044238642
##########
flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/SimpleRuntimeSource.java:
##########
@@ -27,12 +27,14 @@
import javax.annotation.Nullable;
-/** A {@link FlinkSource} for metadata table. */
-public class MetadataSource extends FlinkSource {
+import java.util.Collection;
+
+/** A {@link FlinkSource} for system table. */
+public class SimpleRuntimeSource extends FlinkSource {
private static final long serialVersionUID = 1L;
- public MetadataSource(
+ public SimpleRuntimeSource(
Review Comment:
Very strange name. Readers cannot understand what this class does. Change it. Maybe `SimpleSystemSource`?
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/table/system/AuditLogTable.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.system;
+
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.store.CoreOptions;
+import org.apache.flink.table.store.file.operation.ScanKind;
+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.predicate.PredicateReplaceVisitor;
+import org.apache.flink.table.store.file.utils.RecordReader;
+import org.apache.flink.table.store.file.utils.SnapshotManager;
+import org.apache.flink.table.store.table.DataTable;
+import org.apache.flink.table.store.table.FileStoreTable;
+import org.apache.flink.table.store.table.Table;
+import org.apache.flink.table.store.table.source.DataTableScan;
+import org.apache.flink.table.store.table.source.Split;
+import org.apache.flink.table.store.table.source.TableRead;
+import org.apache.flink.table.store.utils.MappingRowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.RowType.RowField;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.types.RowKind;
+
+import org.apache.flink.shaded.guava30.com.google.common.primitives.Ints;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.store.file.catalog.Catalog.SYSTEM_TABLE_SPLITTER;
+import static org.apache.flink.table.store.file.utils.RecordReaderUtils.transform;
+
+/** A {@link Table} for reading audit log of table. */
+public class AuditLogTable implements DataTable {
+
+ public static final String AUDIT_LOG = "audit_log";
+
+ public static final String ROW_KIND = "rowkind";
+
+ public static final PredicateReplaceVisitor PREDICATE_CONVERTER =
+ p -> {
+ if (p.index() == 0) {
+ return Optional.empty();
+ }
+ return Optional.of(
+ new LeafPredicate(
+ p.function(),
+ p.type(),
+ p.index() - 1,
+ p.fieldName(),
+ p.literals()));
+ };
+
+ private final FileStoreTable dataTable;
+
+ public AuditLogTable(FileStoreTable dataTable) {
+ this.dataTable = dataTable;
+ }
+
+ @Override
+ public String name() {
+ return dataTable.name() + SYSTEM_TABLE_SPLITTER + AUDIT_LOG;
+ }
+
+ @Override
+ public RowType rowType() {
+ List<RowField> fields = new ArrayList<>();
+ fields.add(new RowField(ROW_KIND, new VarCharType(VarCharType.MAX_LENGTH)));
+ fields.addAll(dataTable.rowType().getFields());
+ return new RowType(fields);
+ }
+
+ @Override
+ public DataTableScan newScan() {
+ return new AuditLogScan(dataTable.newScan());
+ }
+
+ @Override
+ public CoreOptions options() {
+ return dataTable.options();
+ }
+
+ @Override
+ public Path location() {
+ return dataTable.location();
+ }
+
+ @Override
+ public SnapshotManager snapshotManager() {
+ return dataTable.snapshotManager();
+ }
+
+ @Override
+ public TableRead newRead() {
+ return new AuditLogRead(dataTable.newRead());
+ }
+
+ @Override
+ public Table copy(Map<String, String> dynamicOptions) {
+ return new AuditLogTable(dataTable.copy(dynamicOptions));
+ }
+
+ /** Push down predicate to dataScan and dataRead. */
+ private Optional<Predicate> convert(Predicate predicate) {
+ List<Predicate> result =
+ PredicateBuilder.splitAnd(predicate).stream()
+ .map(p -> p.visit(PREDICATE_CONVERTER))
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .collect(Collectors.toList());
+ if (result.isEmpty()) {
+ return Optional.empty();
+ }
+ return Optional.of(PredicateBuilder.and(result));
+ }
+
+ private class AuditLogScan implements DataTableScan {
+
+ private final DataTableScan dataScan;
+
+ private AuditLogScan(DataTableScan dataScan) {
+ this.dataScan = dataScan;
+ }
+
+ @Override
+ public DataTableScan withFilter(Predicate predicate) {
+ convert(predicate).ifPresent(dataScan::withFilter);
+ return this;
+ }
+
+ @Override
+ public DataTableScan withKind(ScanKind kind) {
+ dataScan.withKind(kind);
+ return this;
+ }
+
+ @Override
+ public DataTableScan withSnapshot(long snapshotId) {
+ dataScan.withSnapshot(snapshotId);
+ return this;
+ }
+
+ @Override
+ public DataTableScan withLevel(int level) {
+ dataScan.withLevel(level);
+ return this;
+ }
+
+ @Override
+ public DataTableScan.DataFilePlan plan() {
+ return dataScan.plan();
+ }
+ }
+
+ private class AuditLogRead implements TableRead {
+
+ private final TableRead dataRead;
+
+ private int[] readProjection;
+
+ private AuditLogRead(TableRead dataRead) {
+ this.dataRead = dataRead;
+ this.readProjection = defaultProjection();
+ }
+
+ /** Default projection, just add row kind to the first. */
+ private int[] defaultProjection() {
+ int dataFieldCount = dataTable.rowType().getFieldCount();
+ int[] projection = new int[dataFieldCount + 1];
+ projection[0] = -1;
+ for (int i = 0; i < dataFieldCount; i++) {
+ projection[i + 1] = i;
+ }
+ return projection;
+ }
+
+ @Override
+ public TableRead withFilter(Predicate predicate) {
+ convert(predicate).ifPresent(dataRead::withFilter);
+ return this;
+ }
+
+ @Override
+ public TableRead withProjection(int[][] projection) {
+ // data projection to push down to dataRead
+ List<int[]> dataProjection = new ArrayList<>();
+ // read projection to handle record returned by dataRead
+ List<Integer> readProjection = new ArrayList<>();
+ boolean rowKindAppeared = false;
+ for (int i = 0; i < projection.length; i++) {
+ int[] field = projection[i];
+ int topField = field[0];
+ if (topField == 0) {
+ rowKindAppeared = true;
+ readProjection.add(-1);
+ } else {
+ int[] newField = Arrays.copyOf(field, field.length);
+ newField[0] = newField[0] - 1;
+ dataProjection.add(newField);
+
+ // There is no row kind field. Keep it as it is
+ // Row kind field has occurred, and the following fields are offset by 1
+ // position
+ readProjection.add(rowKindAppeared ? i - 1 : i);
+ }
+ }
+ this.readProjection = Ints.toArray(readProjection);
+ dataRead.withProjection(dataProjection.toArray(new int[0][]));
+ return this;
+ }
+
+ @Override
+ public RecordReader<RowData> createReader(Split split) throws IOException {
+ return transform(dataRead.createReader(split), this::convertRow);
+ }
+
+ private RowData convertRow(RowData data) {
+ return new AuditLogRowData(readProjection, data);
+ }
+ }
+
+ /** A {@link MappingRowData} which returns row kind when mapping index is negative. */
+ private static class AuditLogRowData extends MappingRowData {
Review Comment:
No need to specially introduce a `MappingRowData`. Just keep the old `ProjectedRowData` and let this class extend `ProjectedRowData`.
##########
flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/ContinuousFileStoreSource.java:
##########
@@ -36,18 +36,15 @@ public class ContinuousFileStoreSource extends FlinkSource {
private static final long serialVersionUID = 1L;
Review Comment:
Change `serialVersionUID`?
--
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