[jira] [Commented] (DRILL-7273) Create operator for handling metadata

["ASF GitHub Bot (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/DRILL-7273?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16971669#comment-16971669 ] 

ASF GitHub Bot commented on DRILL-7273:
---------------------------------------

vvysotskyi commented on pull request #1886: DRILL-7273: Introduce operators for handling metadata
URL: https://github.com/apache/drill/pull/1886#discussion_r344250388
 
 

 ##########
 File path: exec/java-exec/src/main/java/org/apache/drill/exec/metastore/analyze/FileMetadataInfoCollector.java
 ##########
 @@ -0,0 +1,411 @@
+/*
+ * 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.drill.exec.metastore.analyze;
+
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.physical.base.SchemalessScan;
+import org.apache.drill.exec.planner.FileSystemPartitionDescriptor;
+import org.apache.drill.exec.planner.PartitionLocation;
+import org.apache.drill.exec.planner.logical.DrillRel;
+import org.apache.drill.exec.planner.logical.DrillScanRel;
+import org.apache.drill.exec.planner.logical.DrillTable;
+import org.apache.drill.exec.planner.physical.PlannerSettings;
+import org.apache.drill.exec.store.ColumnExplorer;
+import org.apache.drill.exec.store.dfs.DrillFileSystem;
+import org.apache.drill.exec.store.dfs.FileSelection;
+import org.apache.drill.exec.store.dfs.FormatSelection;
+import org.apache.drill.exec.util.DrillFileSystemUtil;
+import org.apache.drill.exec.util.ImpersonationUtil;
+import org.apache.drill.metastore.components.tables.BasicTablesRequests;
+import org.apache.drill.metastore.metadata.MetadataInfo;
+import org.apache.drill.metastore.metadata.MetadataType;
+import org.apache.drill.metastore.metadata.TableInfo;
+import org.apache.drill.metastore.statistics.TableStatisticsKind;
+import org.apache.drill.shaded.guava.com.google.common.collect.ArrayListMultimap;
+import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
+import org.apache.drill.shaded.guava.com.google.common.collect.Multimap;
+import org.apache.drill.shaded.guava.com.google.common.collect.Streams;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+/**
+ * Implementation of {@link MetadataInfoCollector} for file-based tables.
+ */
+public class FileMetadataInfoCollector implements MetadataInfoCollector {
+  private final List<MetadataInfo> allMetaToHandle;
+  private final List<MetadataInfo> metadataToRemove;
+
+  private final BasicTablesRequests basicRequests;
+  private final TableInfo tableInfo;
+  private final MetadataType metadataLevel;
+
+  private List<MetadataInfo> rowGroupsInfo = Collections.emptyList();
+  private List<MetadataInfo> filesInfo = Collections.emptyList();
+  private Multimap<Integer, MetadataInfo> segmentsInfo = ArrayListMultimap.create();
+
+  private TableScan tableScan;
+
+  private boolean isChanged = true;
+
+  public FileMetadataInfoCollector(BasicTablesRequests basicRequests, TableInfo tableInfo, FormatSelection selection,
+      PlannerSettings settings, Supplier<TableScan> tableScanSupplier, List<SchemaPath> interestingColumns,
+      MetadataType metadataLevel, int segmentColumnsCount) throws IOException {
+    this.basicRequests = basicRequests;
+    this.tableInfo = tableInfo;
+    this.metadataLevel = metadataLevel;
+    this.allMetaToHandle = new ArrayList<>();
+    this.metadataToRemove = new ArrayList<>();
+    init(selection, settings, tableScanSupplier, interestingColumns, segmentColumnsCount);
+  }
+
+  @Override
+  public List<MetadataInfo> getRowGroupsInfo() {
+    return rowGroupsInfo;
+  }
+
+  @Override
+  public List<MetadataInfo> getFilesInfo() {
+    return filesInfo;
+  }
+
+  @Override
+  public Multimap<Integer, MetadataInfo> getSegmentsInfo() {
+    return segmentsInfo;
+  }
+
+  @Override
+  public List<MetadataInfo> getAllMetaToHandle() {
+    return allMetaToHandle;
+  }
+
+  @Override
+  public List<MetadataInfo> getMetadataToRemove() {
+    return metadataToRemove;
+  }
+
+  @Override
+  public TableScan getPrunedScan() {
+    return tableScan;
+  }
+
+  @Override
+  public boolean isChanged() {
+    return isChanged;
+  }
+
+  private void init(FormatSelection selection, PlannerSettings settings, Supplier<TableScan> tableScanSupplier,
+      List<SchemaPath> interestingColumns, int segmentColumnsCount) throws IOException {
+    List<SchemaPath> metastoreInterestingColumns =
+        Optional.ofNullable(basicRequests.interestingColumnsAndPartitionKeys(tableInfo).interestingColumns())
+            .map(metastoreInterestingColumnNames -> metastoreInterestingColumnNames.stream()
+                .map(SchemaPath::parseFromString)
+                .collect(Collectors.toList()))
+        .orElse(null);
+
+    Map<String, Long> filesNamesLastModifiedTime = basicRequests.filesLastModifiedTime(tableInfo, null, null);
+
+    List<String> newFiles = new ArrayList<>();
+    List<String> updatedFiles = new ArrayList<>();
+    List<String> removedFiles = new ArrayList<>(filesNamesLastModifiedTime.keySet());
+    List<String> allFiles = new ArrayList<>();
+
+    for (FileStatus fileStatus : getFileStatuses(selection)) {
+      String path = Path.getPathWithoutSchemeAndAuthority(fileStatus.getPath()).toUri().getPath();
+      Long lastModificationTime = filesNamesLastModifiedTime.get(path);
+      if (lastModificationTime == null) {
+        newFiles.add(path);
+      } else if (lastModificationTime < fileStatus.getModificationTime()) {
+        updatedFiles.add(path);
+      }
+      removedFiles.remove(path);
+      allFiles.add(path);
+    }
+
+    String selectionRoot = selection.getSelection().getSelectionRoot().toUri().getPath();
+
+    if (!Objects.equals(metastoreInterestingColumns, interestingColumns)
+        && (metastoreInterestingColumns == null || !metastoreInterestingColumns.containsAll(interestingColumns))
+        || TableStatisticsKind.ANALYZE_METADATA_LEVEL.getValue(basicRequests.tableMetadata(tableInfo)).compareTo(metadataLevel) != 0) {
+      // do not update table scan and lists of segments / files / row groups,
+      // metadata should be recalculated
+      tableScan = tableScanSupplier.get();
+      metadataToRemove.addAll(getMetadataInfoList(selectionRoot, removedFiles, MetadataType.SEGMENT, 0));
+      return;
+    }
+
+    // checks whether there are no new, updated and removed files
+    if (!newFiles.isEmpty() || !updatedFiles.isEmpty() || !removedFiles.isEmpty()) {
+      List<String> scanFiles = new ArrayList<>(newFiles);
+      scanFiles.addAll(updatedFiles);
+
+      // updates scan to read updated / new files
+      tableScan = getTableScan(settings, tableScanSupplier.get(), scanFiles);
+
+      // iterates from the end;
+      // takes deepest updated segments;
+      // finds their parents:
+      //  - fetches all segments for parent level;
+      //  - filters segments to leave parents only;
+      // obtains all child segments;
+      // filters child segments for filtered parent segments
+
+      int lastSegmentIndex = segmentColumnsCount - 1;
+
+      List<String> scanAndRemovedFiles = new ArrayList<>(scanFiles);
+      scanAndRemovedFiles.addAll(removedFiles);
+
+      // 1. Obtain files info for files from the same folder without removed files
+      // 2. Get segments for obtained files + segments for removed files
+      // 3. Get parent segments
+      // 4. Get other segments for the same parent segment
+      // 5. Remove segments which have only removed files (matched for removedFileInfo and don't match to filesInfo)
+      // 6. Do the same for parent segments
+
+      List<MetadataInfo> allFilesInfo = getMetadataInfoList(selectionRoot, allFiles, MetadataType.FILE, 0);
+
+      // first pass: collect updated segments even without files, they will be removed later
+      List<MetadataInfo> leafSegments = getMetadataInfoList(selectionRoot, scanAndRemovedFiles, MetadataType.SEGMENT, lastSegmentIndex);
+      List<MetadataInfo> removedFilesMetadata = getMetadataInfoList(selectionRoot, removedFiles, MetadataType.FILE, 0);
+
+      List<MetadataInfo> scanFilesInfo = getMetadataInfoList(selectionRoot, scanAndRemovedFiles, MetadataType.FILE, 0);
+      // files from scan + files from the same folder without removed files
+      filesInfo = leafSegments.stream()
+          .filter(parent -> scanFilesInfo.stream().anyMatch(child -> MetadataIdentifierUtils.isMetadataKeyParent(parent.identifier(), child.identifier())))
+          .flatMap(parent ->
+              allFilesInfo.stream()
+                  .filter(child -> MetadataIdentifierUtils.isMetadataKeyParent(parent.identifier(), child.identifier())))
+          .collect(Collectors.toList());
+
+      Multimap<Integer, MetadataInfo> allSegments = populateSegments(removedFiles, allFiles, selectionRoot, lastSegmentIndex, leafSegments, removedFilesMetadata);
+
+      List<MetadataInfo> allRowGroupsInfo = getAllRowGroupsMetadataInfos(allFiles);
+
+      rowGroupsInfo = allRowGroupsInfo.stream()
+          .filter(child -> filesInfo.stream()
+              .map(MetadataInfo::identifier)
+              .anyMatch(parent -> MetadataIdentifierUtils.isMetadataKeyParent(parent, child.identifier())))
+          .collect(Collectors.toList());
+
+      List<MetadataInfo> segmentsToUpdate = getMetadataInfoList(selectionRoot, scanAndRemovedFiles, MetadataType.SEGMENT, 0);
+      Streams.concat(allSegments.values().stream(), allFilesInfo.stream(), allRowGroupsInfo.stream())
+          .filter(child -> segmentsToUpdate.stream().anyMatch(parent -> MetadataIdentifierUtils.isMetadataKeyParent(parent.identifier(), child.identifier())))
+          .filter(parent ->
+              removedFilesMetadata.stream().noneMatch(child -> MetadataIdentifierUtils.isMetadataKeyParent(parent.identifier(), child.identifier()))
+                  || filesInfo.stream().anyMatch(child -> MetadataIdentifierUtils.isMetadataKeyParent(parent.identifier(), child.identifier())))
+          .forEach(allMetaToHandle::add);
+
+      allMetaToHandle.addAll(segmentsToUpdate);
+
+      // is handled separately since it is not overridden when writing the metadata
 
 Review comment:
   It was meant about removed top-level segments, thanks, updated comment.
 
----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


> Create operator for handling metadata
> -------------------------------------
>
>                 Key: DRILL-7273
>                 URL: https://issues.apache.org/jira/browse/DRILL-7273
>             Project: Apache Drill
>          Issue Type: Sub-task
>            Reporter: Arina Ielchiieva
>            Assignee: Vova Vysotskyi
>            Priority: Major
>              Labels: doc-impacting
>             Fix For: 1.17.0
>
>
> As described in the design document for [File metastore table metadata|https://docs.google.com/document/d/14pSIzKqDltjLEEpEebwmKnsDPxyS_6jGrPOjXu6M_NM/edit#heading=h.j079wdhtehuk] it is required to create an operator for handling metadata to be able to do incremental analyze.
> Add options:
> metastore.enabled 
> metastore.metadata.store.depth_level
> Statement:
> ANALYZE TABLE REFRESH METADATA ...



--
This message was sent by Atlassian Jira
(v8.3.4#803005)