LuciferYang commented on a change in pull request #30483:
URL: https://github.com/apache/spark/pull/30483#discussion_r691809788
##########
File path:
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/SpecificParquetRecordReaderBase.java
##########
@@ -77,28 +82,31 @@
protected ParquetFileReader reader;
+ protected ParquetMetadata cachedFooter;
+
@Override
public void initialize(InputSplit inputSplit, TaskAttemptContext
taskAttemptContext)
throws IOException, InterruptedException {
Configuration configuration = taskAttemptContext.getConfiguration();
FileSplit split = (FileSplit) inputSplit;
this.file = split.getPath();
- ParquetReadOptions options = HadoopReadOptions
- .builder(configuration)
- .withRange(split.getStart(), split.getStart() + split.getLength())
- .build();
- this.reader = new ParquetFileReader(HadoopInputFile.fromPath(file,
configuration), options);
- this.fileSchema = reader.getFileMetaData().getSchema();
- Map<String, String> fileMetadata =
reader.getFileMetaData().getKeyValueMetaData();
+ ParquetMetadata footer =
+ readFooterByRange(configuration, split.getStart(), split.getStart() +
split.getLength());
+ this.fileSchema = footer.getFileMetaData().getSchema();
+ FilterCompat.Filter filter = ParquetInputFormat.getFilter(configuration);
+ List<BlockMetaData> blocks =
+ RowGroupFilter.filterRowGroups(filter, footer.getBlocks(), fileSchema);
Review comment:
I need to investigate it again
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileMetaCacheManager.scala
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.spark.sql.execution.datasources
+
+import java.util.concurrent.TimeUnit
+
+import com.github.benmanes.caffeine.cache.{CacheLoader, Caffeine}
+import com.github.benmanes.caffeine.cache.stats.CacheStats
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+
+import org.apache.spark.SparkEnv
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * A singleton Cache Manager to caching file meta. We cache these file metas
in order to speed up
+ * iterated queries over the same dataset. Otherwise, each query would have to
hit remote storage
+ * in order to fetch file meta before read files.
+ *
+ * We should implement the corresponding `FileMetaKey` for a specific file
format, for example
+ * `ParquetFileMetaKey` or `OrcFileMetaKey`. By default, the file path is used
as the identification
+ * of the `FileMetaKey` and the `getFileMeta` method of `FileMetaKey` is used
to return the file
+ * meta of the corresponding file format.
+ */
+object FileMetaCacheManager extends Logging {
+
+ private lazy val cacheLoader = new CacheLoader[FileMetaKey, FileMeta]() {
+ override def load(entry: FileMetaKey): FileMeta = {
+ logDebug(s"Loading Data File Meta ${entry.path}")
+ entry.getFileMeta
+ }
+ }
+
+ private lazy val ttlTime =
+ SparkEnv.get.conf.get(SQLConf.FILE_META_CACHE_TTL_SINCE_LAST_ACCESS)
+
+ private lazy val cache = Caffeine
+ .newBuilder()
+ .expireAfterAccess(ttlTime, TimeUnit.SECONDS)
+ .recordStats()
+ .build[FileMetaKey, FileMeta](cacheLoader)
+
+ /**
+ * Returns the `FileMeta` associated with the `FileMetaKey` in the
`FileMetaCacheManager`,
+ * obtaining that the `FileMeta` from `cacheLoader.load(FileMetaKey)` if
necessary.
+ */
+ def get(dataFile: FileMetaKey): FileMeta = cache.get(dataFile)
+
+ /**
+ * This is visible for testing.
+ */
+ def cacheStats: CacheStats = cache.stats()
+
+ /**
+ * This is visible for testing.
+ */
+ def cleanUp(): Unit = cache.cleanUp()
+}
+
+abstract class FileMetaKey {
+ def path: Path
+ def configuration: Configuration
+ def getFileMeta: FileMeta
+ override def hashCode(): Int = path.hashCode
+ override def equals(other: Any): Boolean = other match {
+ case df: FileMetaKey => path.equals(df.path)
Review comment:
This is a very good question, we discussed in
https://github.com/apache/spark/pull/33748#discussion_r689993828,
```
If the file name has the timestamp, I think we don't have to worry too much.
The names of the new file and the old file are different and they can ensure
that they don't read the wrong data.
If it is manually file replaced and the file has the same name and the
corresponding file meta exists in the cache, an incorrect file meta will be
used to read the data. If the data reading fails, the job will fail. But if the
data reading happens to be successful, the job will read the wrong data.
In fact, even if there is no `FileMetaCache`, there is a similar risk in
manually replace files with same name, because the offset and length of
PartitionedFile maybe don't match after manually replace for a running job
```
And At the same time, I added a warning for this feature in SQLConf.
Now Parquet is a draft because the Deprecated API, We are focusing on ORC
(SPARK-36516) now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
