You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by ih...@apache.org on 2019/08/05 07:38:48 UTC
[drill] branch master updated (5df4495 -> 7bd442d)
This is an automated email from the ASF dual-hosted git repository.
ihuzenko pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git.
from 5df4495 DRILL-7327: Log Regex Plugin Won't Recognize Schema
new 22947ad DRILL-7084: ResultSet getObject method throws not implemented exception if the column type is NULL
new fc3d921 DRILL-7331: Drill Iceberg Metastore metadata expiration
new df48eb8 DRILL-7334: Update Iceberg Metastore Parquet write mode
new 7bd442d DRILL-7335: Fix error when reading csv file with headers only
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/drill/exec/server/DrillbitContext.java | 1 +
.../exec/store/easy/text/reader/TextInput.java | 10 +-
.../easy/text/compliant/TestCsvWithHeaders.java | 62 +++--
.../apache/drill/jdbc/impl/DrillResultSetImpl.java | 24 +-
.../org/apache/drill/jdbc/DrillResultSetTest.java | 109 ++++----
.../src/test/resources/testGetObjectNull.parquet | Bin 0 -> 379 bytes
metastore/iceberg-metastore/README.md | 44 ++--
metastore/iceberg-metastore/pom.xml | 2 +-
.../drill/metastore/iceberg/IcebergMetastore.java | 29 ++-
.../drill/metastore/iceberg/MetastoreContext.java | 8 +
.../iceberg/components/tables/IcebergTables.java | 10 +-
.../iceberg/config/IcebergConfigConstants.java | 5 +
.../iceberg/operate/ExpirationHandler.java | 276 +++++++++++++++++++++
.../metastore/iceberg/operate/IcebergModify.java | 3 +
.../metastore/iceberg/write/ParquetFileWriter.java | 7 +-
.../src/main/resources/drill-metastore-module.conf | 65 ++---
.../iceberg/components/tables/TestTableKey.java | 2 +-
.../tables/TestTablesOperationTransformer.java | 5 +-
.../iceberg/operate/TestExpirationHandler.java | 143 +++++++++++
.../iceberg/write/TestParquetFileWriter.java | 8 +-
.../java/org/apache/drill/metastore/Metastore.java | 2 +-
.../apache/drill/metastore/MetastoreRegistry.java | 9 +-
.../main/resources/drill-metastore-default.conf | 2 +-
23 files changed, 659 insertions(+), 167 deletions(-)
create mode 100644 exec/jdbc/src/test/resources/testGetObjectNull.parquet
create mode 100644 metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/operate/ExpirationHandler.java
create mode 100644 metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/operate/TestExpirationHandler.java
[drill] 04/04: DRILL-7335: Fix error when reading csv file with
headers only
Posted by ih...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ihuzenko pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit 7bd442d02093240bc31aca827d57e12eb4c2f4f5
Author: Arina Ielchiieva <ar...@gmail.com>
AuthorDate: Thu Aug 1 20:18:34 2019 +0300
DRILL-7335: Fix error when reading csv file with headers only
closes #1834
---
.../exec/store/easy/text/reader/TextInput.java | 10 ++--
.../easy/text/compliant/TestCsvWithHeaders.java | 62 ++++++++++++++--------
2 files changed, 46 insertions(+), 26 deletions(-)
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/reader/TextInput.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/reader/TextInput.java
index 3e05e58..d9fa973 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/reader/TextInput.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/reader/TextInput.java
@@ -86,9 +86,9 @@ final class TextInput {
/**
* Creates a new instance with the mandatory characters for handling newlines
* transparently. lineSeparator the sequence of characters that represent a
- * newline, as defined in {@link Format#getLineSeparator()}
+ * newline, as defined in {@link TextParsingSettings#getNewLineDelimiter()}
* normalizedLineSeparator the normalized newline character (as defined in
- * {@link Format#getNormalizedNewline()}) that is used to replace any
+ * {@link TextParsingSettings#getNormalizedNewLine()}) that is used to replace any
* lineSeparator sequence found in the input.
*/
public TextInput(TextParsingSettings settings, InputStream input, DrillBuf readBuffer, long startPos, long endPos) {
@@ -142,7 +142,11 @@ final class TextInput {
if (startPos > 0 || settings.isSkipFirstLine()) {
// move to next full record.
- skipLines(1);
+ try {
+ skipLines(1);
+ } catch (StreamFinishedPseudoException e) {
+ // file does not have any more lines, ignore
+ }
}
}
}
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithHeaders.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithHeaders.java
index 8a2e8b0..4d4202c 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithHeaders.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithHeaders.java
@@ -23,7 +23,10 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import java.io.File;
+import java.io.FileWriter;
import java.io.IOException;
+import java.io.PrintWriter;
import java.util.Iterator;
import org.apache.drill.categories.RowSetTests;
@@ -55,34 +58,34 @@ import org.junit.experimental.categories.Category;
* the first batch either contains data, or that the first batch is empty
* only if there is no data at all to be read.
*
- * @see {@link TestHeaderBuilder}
+ * @see TestHeaderBuilder
*/
@Category(RowSetTests.class)
public class TestCsvWithHeaders extends BaseCsvTest {
private static final String TEST_FILE_NAME = "basic.csv";
+ private static final String COLUMNS_FILE_NAME = "columns.csv";
+ private static final String EMPTY_HEADERS_FILE = "noHeaders.csv";
- private static String invalidHeaders[] = {
+ private static String[] invalidHeaders = {
"$,,9b,c,c,c_2",
"10,foo,bar,fourth,fifth,sixth"
};
- private static String emptyHeaders[] = {
+ private static String[] emptyHeaders = {
"",
"10,foo,bar"
};
- private static String raggedRows[] = {
+ private static String[] raggedRows = {
"a,b,c",
"10,dino",
"20,foo,bar",
"30"
};
- public static final String COLUMNS_FILE_NAME = "columns.csv";
-
- private static String columnsCol[] = {
+ private static String[] columnsCol = {
"author,columns",
"fred,\"Rocks Today,Dino Wrangling\"",
"barney,Bowlarama"
@@ -109,7 +112,7 @@ public class TestCsvWithHeaders extends BaseCsvTest {
* <br><tt>SELECT * FROM VALUES ();</tt><br>
* The implementation tested here follows that pattern.
*
- * @see {@link TestCsvWithoutHeaders#testEmptyFile()}
+ * @see TestCsvWithoutHeaders#testEmptyFile()
*/
@Test
public void testEmptyFile() throws IOException {
@@ -118,8 +121,6 @@ public class TestCsvWithHeaders extends BaseCsvTest {
assertNull(rowSet);
}
- private static final String EMPTY_HEADERS_FILE = "noheaders.csv";
-
/**
* Trivial case: empty header. This case should fail.
*/
@@ -188,10 +189,6 @@ public class TestCsvWithHeaders extends BaseCsvTest {
RowSetUtilities.verify(expected, actual);
}
- private String makeStatement(String fileName) {
- return "SELECT * FROM `dfs.data`.`" + fileName + "`";
- }
-
/**
* Verify that the wildcard expands columns to the header names, including
* case
@@ -352,9 +349,8 @@ public class TestCsvWithHeaders extends BaseCsvTest {
* files are nested to another level.)
*/
@Test
- public void testPartitionExpansion() throws IOException {
- String sql = "SELECT * FROM `dfs.data`.`%s`";
- Iterator<DirectRowSet> iter = client.queryBuilder().sql(sql, PART_DIR).rowSetIterator();
+ public void testPartitionExpansion() {
+ Iterator<DirectRowSet> iter = client.queryBuilder().sql(makeStatement(PART_DIR)).rowSetIterator();
TupleMetadata expectedSchema = new SchemaBuilder()
.add("a", MinorType.VARCHAR)
@@ -407,7 +403,7 @@ public class TestCsvWithHeaders extends BaseCsvTest {
* partition column moves after data columns.
*/
@Test
- public void testWilcardAndPartitionsMultiFiles() throws IOException {
+ public void testWildcardAndPartitionsMultiFiles() {
String sql = "SELECT *, dir0, dir1 FROM `dfs.data`.`%s`";
Iterator<DirectRowSet> iter = client.queryBuilder().sql(sql, PART_DIR).rowSetIterator();
@@ -464,7 +460,7 @@ public class TestCsvWithHeaders extends BaseCsvTest {
* are consistent even when used across multiple scans.
*/
@Test
- public void doTestExplicitPartitionsMultiFiles() throws IOException {
+ public void doTestExplicitPartitionsMultiFiles() {
String sql = "SELECT a, b, c, dir0, dir1 FROM `dfs.data`.`%s`";
Iterator<DirectRowSet> iter = client.queryBuilder().sql(sql, PART_DIR).rowSetIterator();
@@ -537,7 +533,6 @@ public class TestCsvWithHeaders extends BaseCsvTest {
* The column name `columns` is treated as a plain old
* column when using column headers. If used with an index,
* validation will fail because the VarChar column is not an array
- * @throws Exception
*/
@Test
public void testColumnsIndex() throws Exception {
@@ -572,7 +567,6 @@ public class TestCsvWithHeaders extends BaseCsvTest {
/**
* If columns[x] is used, then this can't possibly match a valid
* text reader column, so raise an error instead.
- * @throws Exception
*/
@Test
public void testColumnsIndexMissing() throws Exception {
@@ -594,8 +588,7 @@ public class TestCsvWithHeaders extends BaseCsvTest {
@Test
public void testHugeColumn() throws IOException {
String fileName = buildBigColFile(true);
- String sql = "SELECT * FROM `dfs.data`.`%s`";
- RowSet actual = client.queryBuilder().sql(sql, fileName).rowSet();
+ RowSet actual = client.queryBuilder().sql(makeStatement(fileName)).rowSet();
assertEquals(10, actual.rowCount());
RowSetReader reader = actual.reader();
while (reader.next()) {
@@ -610,4 +603,27 @@ public class TestCsvWithHeaders extends BaseCsvTest {
}
actual.clear();
}
+
+ @Test
+ public void testHeadersOnly() throws Exception {
+ String fileName = "headersOnly.csv";
+ try (PrintWriter out = new PrintWriter(new FileWriter(new File(testDir, fileName)))) {
+ out.print("a,b,c"); // note: no \n in the end
+ }
+
+ RowSet actual = client.queryBuilder().sql(makeStatement(fileName)).rowSet();
+
+ TupleMetadata expectedSchema = new SchemaBuilder()
+ .add("a", MinorType.VARCHAR)
+ .add("b", MinorType.VARCHAR)
+ .add("c", MinorType.VARCHAR)
+ .buildSchema();
+ RowSet expected = new RowSetBuilder(client.allocator(), expectedSchema)
+ .build();
+ RowSetUtilities.verify(expected, actual);
+ }
+
+ private String makeStatement(String fileName) {
+ return "SELECT * FROM `dfs.data`.`" + fileName + "`";
+ }
}
[drill] 03/04: DRILL-7334: Update Iceberg Metastore Parquet write
mode
Posted by ih...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ihuzenko pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit df48eb8167219b7c632e8f6d760e76242805d6af
Author: Arina Ielchiieva <ar...@gmail.com>
AuthorDate: Wed Jul 31 14:04:32 2019 +0300
DRILL-7334: Update Iceberg Metastore Parquet write mode
closes #1832
---
metastore/iceberg-metastore/README.md | 19 ++++++++++---------
metastore/iceberg-metastore/pom.xml | 2 +-
.../metastore/iceberg/write/ParquetFileWriter.java | 7 ++++---
.../src/main/resources/drill-metastore-module.conf | 9 ++++-----
.../iceberg/write/TestParquetFileWriter.java | 8 +++-----
5 files changed, 22 insertions(+), 23 deletions(-)
diff --git a/metastore/iceberg-metastore/README.md b/metastore/iceberg-metastore/README.md
index cc46c2a..3db08d1 100644
--- a/metastore/iceberg-metastore/README.md
+++ b/metastore/iceberg-metastore/README.md
@@ -56,16 +56,18 @@ Assume Iceberg table location is `/drill/metastore/iceberg/tables`, metadata for
### Metadata Storage Format
-By default, Metadata will be stored in Parquet files.
-Each parquet file will hold information for one partition.
+Iceberg tables support data storage in three formats: Parquet, Avro, ORC.
+Drill metadata will be stored in Parquet files. This format was chosen over others
+since it is column oriented and efficient in terms of disk I/O
+when specific columns need to be queried.
+
+Each Parquet file will hold information for one partition.
Partition keys will depend on Metastore component characteristics.
For example, for `tables` component, partitions keys are
storage plugin, workspace, table name and metadata key.
Parquet files name will be based on `UUID` to ensure uniqueness.
-
-Iceberg also supports data storage in Avro and ORC files, writing metadata
-in these formats can be added later.
+If somehow collision occurs, modify operation in Metastore will fail.
## Metastore Operations flow
@@ -73,8 +75,7 @@ Metastore main goal is to provide ability to read and modify metadata.
### Read
-Metastore data is read using `IcebergGenerics#read`. Iceberg will automatically determine
-format in which data is stored (three formats are supported Parquet, Avro, ORC).
+Metastore data is read using `IcebergGenerics#read`.
Based on given filter and select columns list, data will be returned in
`org.apache.iceberg.data.Record` format which will be transformed
into the list of Metastore component units and returned to the caller.
@@ -86,7 +87,7 @@ partition keys can be included into filter expression.
### Add
To add metadata to Iceberg table, caller provides list of component units which
-will be written into Parquet files (current default format) and grouped by partition keys.
+will be written into Parquet files and grouped by partition keys.
Each group will be written into separate Parquet file
and stored in the location inside of Iceberg table based on component unit location keys.
Note: partition keys must not be null.
@@ -107,7 +108,7 @@ Parquet files with metadata for this table will be stored in
`[METASTORE_ROOT_DIRECTORY]/[COMPONENT_LOCATION]/dfs/tmp/nation` folder.
If `dfs.tmp.nation` is un-partitioned, it's metadata will be stored in two
-parquet files: one file with general table information,
+Parquet files: one file with general table information,
another file with default segment information.
If `dfs.tmp.nation` is partitioned, it will have also one file with general
information and `N` files with top-level segments information.
diff --git a/metastore/iceberg-metastore/pom.xml b/metastore/iceberg-metastore/pom.xml
index f8544fd..d935750 100644
--- a/metastore/iceberg-metastore/pom.xml
+++ b/metastore/iceberg-metastore/pom.xml
@@ -33,7 +33,7 @@
<name>metastore/Drill Iceberg Metastore</name>
<properties>
- <iceberg.version>77a456a</iceberg.version>
+ <iceberg.version>08e0873</iceberg.version>
<caffeine.version>2.7.0</caffeine.version>
</properties>
diff --git a/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/write/ParquetFileWriter.java b/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/write/ParquetFileWriter.java
index 3c1604d..3286676 100644
--- a/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/write/ParquetFileWriter.java
+++ b/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/write/ParquetFileWriter.java
@@ -23,6 +23,7 @@ import org.apache.iceberg.FileFormat;
import org.apache.iceberg.Table;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.exceptions.RuntimeIOException;
import org.apache.iceberg.io.FileAppender;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.parquet.Parquet;
@@ -35,7 +36,7 @@ import java.util.Objects;
/**
* Parquet File Writer implementation. Creates Parquet file in given location and name
* and '.parquet' extension and writes given data into it.
- * Note: if file already exists, it will be overwritten.
+ * Note: if file already exists, write operation will fail.
*/
public class ParquetFileWriter implements FileWriter {
@@ -83,7 +84,7 @@ public class ParquetFileWriter implements FileWriter {
fileAppender.close();
// metrics are available only when file was written (i.e. close method was executed)
return new File(outputFile, fileAppender.metrics());
- } catch (IOException | ClassCastException e) {
+ } catch (IOException | ClassCastException | RuntimeIOException e) {
if (fileAppender != null) {
try {
fileAppender.close();
@@ -91,7 +92,7 @@ public class ParquetFileWriter implements FileWriter {
// write has failed anyway, ignore closing exception if any and throw initial one
}
}
- throw new IcebergMetastoreException("Unable to write data into parquet file", e);
+ throw new IcebergMetastoreException(String.format("Unable to write data into parquet file [%s]", outputFile.location()), e);
}
}
}
diff --git a/metastore/iceberg-metastore/src/main/resources/drill-metastore-module.conf b/metastore/iceberg-metastore/src/main/resources/drill-metastore-module.conf
index be98a97..33fe795 100644
--- a/metastore/iceberg-metastore/src/main/resources/drill-metastore-module.conf
+++ b/metastore/iceberg-metastore/src/main/resources/drill-metastore-module.conf
@@ -43,11 +43,10 @@ drill.metastore.iceberg: {
}
components: {
- // Common properties for all Iceberg tables from org.apache.iceberg.TableProperties can be specified
- common.properties: {
- write.format.default: "parquet",
- write.metadata.compression-codec: "none"
- },
+ // Common properties for all Iceberg tables from org.apache.iceberg.TableProperties can be specified
+ common.properties: {
+ write.metadata.compression-codec: "none"
+ },
tables: {
// Iceberg table location in Iceberg Metastore
diff --git a/metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/write/TestParquetFileWriter.java b/metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/write/TestParquetFileWriter.java
index 6943185..4a9211a 100644
--- a/metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/write/TestParquetFileWriter.java
+++ b/metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/write/TestParquetFileWriter.java
@@ -261,15 +261,13 @@ public class TestParquetFileWriter extends IcebergBaseTest {
java.nio.file.Path file = Paths.get(new File(location, FileFormat.PARQUET.addExtension(fileName)).getPath());
Files.write(file, Collections.singletonList("abc"));
- org.apache.drill.metastore.iceberg.write.File result = new ParquetFileWriter(table)
+ thrown.expect(IcebergMetastoreException.class);
+
+ new ParquetFileWriter(table)
.records(Collections.singletonList(record))
.location(location)
.name(fileName)
.write();
-
- List<Record> rows = readData(result.input(), schema);
- assertEquals(1, rows.size());
- assertEquals(1, rows.get(0).getField("int_field"));
}
private List<Record> readData(InputFile inputFile, Schema schema) throws IOException {
[drill] 01/04: DRILL-7084: ResultSet getObject method throws not
implemented exception if the column type is NULL
Posted by ih...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ihuzenko pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit 22947ad336d8dd0d944a66ded16a7ce9ffa62f04
Author: Anton Gozhiy <an...@gmail.com>
AuthorDate: Tue Jul 9 18:04:52 2019 +0300
DRILL-7084: ResultSet getObject method throws not implemented exception if the column type is NULL
closes #1825
---
.../apache/drill/jdbc/impl/DrillResultSetImpl.java | 24 ++---
.../org/apache/drill/jdbc/DrillResultSetTest.java | 109 ++++++++++-----------
.../src/test/resources/testGetObjectNull.parquet | Bin 0 -> 379 bytes
3 files changed, 65 insertions(+), 68 deletions(-)
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/impl/DrillResultSetImpl.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/impl/DrillResultSetImpl.java
index 5948393..0ce1e79 100644
--- a/exec/jdbc/src/main/java/org/apache/drill/jdbc/impl/DrillResultSetImpl.java
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/impl/DrillResultSetImpl.java
@@ -78,11 +78,11 @@ public class DrillResultSetImpl extends AvaticaResultSet implements DrillResultS
* Throws AlreadyClosedSqlException or QueryCanceledSqlException if this
* ResultSet is closed.
*
- * @throws ExecutionCanceledSqlException if ResultSet is closed because of
- * cancelation and no QueryCanceledSqlException has been thrown yet
- * for this ResultSet
- * @throws AlreadyClosedSqlException if ResultSet is closed
- * @throws SQLException if error in calling {@link #isClosed()}
+ * @throws ExecutionCanceledSqlException if ResultSet is closed because of
+ * cancellation and no QueryCanceledSqlException has been thrown yet
+ * for this ResultSet
+ * @throws AlreadyClosedSqlException if ResultSet is closed
+ * @throws SQLException if error in calling {@link #isClosed()}
*/
@Override
protected void checkOpen() throws SQLException {
@@ -168,7 +168,7 @@ public class DrillResultSetImpl extends AvaticaResultSet implements DrillResultS
}
@Override
- public Object getObject( int columnIndex ) throws SQLException {
+ public Object getObject(int columnIndex) throws SQLException {
checkOpen();
Cursor.Accessor accessor;
@@ -180,7 +180,7 @@ public class DrillResultSetImpl extends AvaticaResultSet implements DrillResultS
ColumnMetaData metaData = columnMetaDataList.get(columnIndex - 1);
// Drill returns a float (4bytes) for a SQL Float whereas Calcite would return a double (8bytes)
int typeId = (metaData.type.id != Types.FLOAT) ? metaData.type.id : Types.REAL;
- return AvaticaSite.get(accessor, typeId, localCalendar);
+ return typeId != Types.NULL ? AvaticaSite.get(accessor, typeId, localCalendar) : null;
}
//--------------------------JDBC 2.0-----------------------------------
@@ -249,10 +249,10 @@ public class DrillResultSetImpl extends AvaticaResultSet implements DrillResultS
}
@Override
- public boolean relative( int rows ) throws SQLException {
+ public boolean relative(int rows) throws SQLException {
checkOpen();
try {
- return super.relative( rows );
+ return super.relative(rows);
} catch (UnsupportedOperationException e) {
throw new SQLFeatureNotSupportedException(e.getMessage(), e);
}
@@ -275,7 +275,7 @@ public class DrillResultSetImpl extends AvaticaResultSet implements DrillResultS
public void updateNull(int columnIndex) throws SQLException {
checkOpen();
try {
- super.updateNull( columnIndex );
+ super.updateNull(columnIndex);
} catch (UnsupportedOperationException e) {
throw new SQLFeatureNotSupportedException(e.getMessage(), e);
}
@@ -295,7 +295,7 @@ public class DrillResultSetImpl extends AvaticaResultSet implements DrillResultS
public void updateByte(int columnIndex, byte x) throws SQLException {
checkOpen();
try {
- super.updateByte( columnIndex, x );
+ super.updateByte(columnIndex, x);
} catch (UnsupportedOperationException e) {
throw new SQLFeatureNotSupportedException(e.getMessage(), e);
}
@@ -1260,7 +1260,7 @@ public class DrillResultSetImpl extends AvaticaResultSet implements DrillResultS
////////////////////////////////////////
@Override
- protected DrillResultSetImpl execute() throws SQLException{
+ protected DrillResultSetImpl execute() throws SQLException {
connection.getDriver().handler.onStatementExecute(statement, null);
if (signature.cursorFactory != null) {
diff --git a/exec/jdbc/src/test/java/org/apache/drill/jdbc/DrillResultSetTest.java b/exec/jdbc/src/test/java/org/apache/drill/jdbc/DrillResultSetTest.java
index e75d19c..73f96e5 100644
--- a/exec/jdbc/src/test/java/org/apache/drill/jdbc/DrillResultSetTest.java
+++ b/exec/jdbc/src/test/java/org/apache/drill/jdbc/DrillResultSetTest.java
@@ -18,11 +18,10 @@
package org.apache.drill.jdbc;
import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.core.StringContains.containsString;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
-import static org.junit.Assert.fail;
import java.sql.Connection;
import java.sql.ResultSet;
@@ -34,8 +33,10 @@ import org.apache.drill.exec.ExecConstants;
import org.apache.drill.categories.JdbcTest;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
+import org.junit.rules.ExpectedException;
@Category({SlowTest.class, JdbcTest.class})
public class DrillResultSetTest extends JdbcTestBase {
@@ -45,123 +46,119 @@ public class DrillResultSetTest extends JdbcTestBase {
ExecConstants.HTTP_ENABLE;
private static final String origStatusServerPropValue =
- System.getProperty( STATUS_SERVER_PROPERTY_NAME, "true" );
+ System.getProperty(STATUS_SERVER_PROPERTY_NAME, "true");
+
+ @Rule
+ public ExpectedException thrown = ExpectedException.none();
// Disable Jetty status server so unit tests run (outside Maven setup).
// (TODO: Move this to base test class and/or have Jetty try other ports.)
@BeforeClass
public static void setUpClass() {
- System.setProperty( STATUS_SERVER_PROPERTY_NAME, "false" );
+ System.setProperty(STATUS_SERVER_PROPERTY_NAME, "false");
}
@AfterClass
public static void tearDownClass() {
- System.setProperty( STATUS_SERVER_PROPERTY_NAME, origStatusServerPropValue );
+ System.setProperty(STATUS_SERVER_PROPERTY_NAME, origStatusServerPropValue);
}
@Test
public void test_next_blocksFurtherAccessAfterEnd()
- throws SQLException
- {
+ throws SQLException {
Connection connection = connect();
Statement statement = connection.createStatement();
ResultSet resultSet =
- statement.executeQuery( "SELECT 1 AS x \n" +
- "FROM cp.`donuts.json` \n" +
- "LIMIT 2" );
+ statement.executeQuery("SELECT 1 AS x \n" +
+ "FROM cp.`donuts.json` \n" +
+ "LIMIT 2");
// Advance to first row; confirm can access data.
- assertThat( resultSet.next(), is( true ) );
- assertThat( resultSet.getInt( 1 ), is ( 1 ) );
+ assertThat(resultSet.next(), is(true));
+ assertThat(resultSet.getInt(1), is(1));
// Advance from first to second (last) row, confirming data access.
- assertThat( resultSet.next(), is( true ) );
- assertThat( resultSet.getInt( 1 ), is ( 1 ) );
+ assertThat(resultSet.next(), is(true));
+ assertThat(resultSet.getInt(1), is(1));
// Now advance past last row.
- assertThat( resultSet.next(), is( false ) );
+ assertThat(resultSet.next(), is(false));
// Main check: That row data access methods now throw SQLException.
- try {
- resultSet.getInt( 1 );
- fail( "Didn't get expected SQLException." );
- }
- catch ( SQLException e ) {
- // Expect something like current InvalidCursorStateSqlException saying
- // "Result set cursor is already positioned past all rows."
- assertThat( e, instanceOf( InvalidCursorStateSqlException.class ) );
- assertThat( e.toString(), containsString( "past" ) );
- }
- // (Any other exception is unexpected result.)
-
- assertThat( resultSet.next(), is( false ) );
+ thrown.expect(InvalidCursorStateSqlException.class);
+ thrown.expectMessage(containsString("Result set cursor is already positioned past all rows."));
+ resultSet.getInt(1);
+
+ assertThat(resultSet.next(), is(false));
// TODO: Ideally, test all other accessor methods.
}
@Test
public void test_next_blocksFurtherAccessWhenNoRows()
- throws Exception
- {
+ throws Exception {
Connection connection = connect();
Statement statement = connection.createStatement();
ResultSet resultSet =
- statement.executeQuery( "SELECT 'Hi' AS x \n" +
- "FROM cp.`donuts.json` \n" +
- "WHERE false" );
+ statement.executeQuery("SELECT 'Hi' AS x \n" +
+ "FROM cp.`donuts.json` \n" +
+ "WHERE false");
// Do initial next(). (Advance from before results to next possible
// position (after the set of zero rows).
- assertThat( resultSet.next(), is( false ) );
+ assertThat(resultSet.next(), is(false));
// Main check: That row data access methods throw SQLException.
- try {
- resultSet.getString( 1 );
- fail( "Didn't get expected SQLException." );
- }
- catch ( SQLException e ) {
- // Expect something like current InvalidRowSQLException saying
- // "Result set cursor is already positioned past all rows."
- assertThat( e, instanceOf( InvalidCursorStateSqlException.class ) );
- assertThat( e.toString(), containsString( "past" ) );
- assertThat( e.toString(), containsString( "rows" ) );
- }
- // (Any non-SQLException exception is unexpected result.)
-
- assertThat( resultSet.next(), is( false ) );
+ thrown.expect(InvalidCursorStateSqlException.class);
+ thrown.expectMessage(containsString("Result set cursor is already positioned past all rows."));
+ resultSet.getString(1);
+
+ assertThat(resultSet.next(), is(false));
// TODO: Ideally, test all other accessor methods.
}
@Test
public void test_getRow_isOneBased()
- throws Exception
- {
+ throws Exception {
Connection connection = connect();
Statement statement = connection.createStatement();
ResultSet resultSet =
- statement.executeQuery( "VALUES (1), (2)" );
+ statement.executeQuery("VALUES (1), (2)");
// Expect 0 when before first row:
- assertThat( "getRow() before first next()", resultSet.getRow(), equalTo( 0 ) );
+ assertThat("getRow() before first next()", resultSet.getRow(), equalTo(0));
resultSet.next();
// Expect 1 at first row:
- assertThat( "getRow() at first row", resultSet.getRow(), equalTo( 1 ) );
+ assertThat("getRow() at first row", resultSet.getRow(), equalTo(1));
resultSet.next();
// Expect 2 at second row:
- assertThat( "getRow() at second row", resultSet.getRow(), equalTo( 2 ) );
+ assertThat("getRow() at second row", resultSet.getRow(), equalTo(2));
resultSet.next();
// Expect 0 again when after last row:
- assertThat( "getRow() after last row", resultSet.getRow(), equalTo( 0 ) );
+ assertThat("getRow() after last row", resultSet.getRow(), equalTo(0));
+ resultSet.next();
+ assertThat("getRow() after last row", resultSet.getRow(), equalTo(0));
+ }
+
+ @Test
+ public void testGetObjectNull() throws SQLException {
+ Connection connection = connect();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet =
+ statement.executeQuery(
+ "select coalesce(a1, b1) " +
+ "from cp.`testGetObjectNull.parquet` " +
+ "limit 1");
resultSet.next();
- assertThat( "getRow() after last row", resultSet.getRow(), equalTo( 0 ) );
+ assertNull(resultSet.getObject(1));
}
// TODO: Ideally, test other methods.
diff --git a/exec/jdbc/src/test/resources/testGetObjectNull.parquet b/exec/jdbc/src/test/resources/testGetObjectNull.parquet
new file mode 100644
index 0000000..f3cfefd
Binary files /dev/null and b/exec/jdbc/src/test/resources/testGetObjectNull.parquet differ
[drill] 02/04: DRILL-7331: Drill Iceberg Metastore metadata
expiration
Posted by ih...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ihuzenko pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit fc3d921f0783cc2044ac4ed1f1a144231b5adbad
Author: Arina Ielchiieva <ar...@gmail.com>
AuthorDate: Thu Jul 25 19:22:47 2019 +0300
DRILL-7331: Drill Iceberg Metastore metadata expiration
closes #1831
---
.../apache/drill/exec/server/DrillbitContext.java | 1 +
metastore/iceberg-metastore/README.md | 25 +-
metastore/iceberg-metastore/pom.xml | 2 +-
.../drill/metastore/iceberg/IcebergMetastore.java | 29 ++-
.../drill/metastore/iceberg/MetastoreContext.java | 8 +
.../iceberg/components/tables/IcebergTables.java | 10 +-
.../iceberg/config/IcebergConfigConstants.java | 5 +
.../iceberg/operate/ExpirationHandler.java | 276 +++++++++++++++++++++
.../metastore/iceberg/operate/IcebergModify.java | 3 +
.../src/main/resources/drill-metastore-module.conf | 74 +++---
.../iceberg/components/tables/TestTableKey.java | 2 +-
.../tables/TestTablesOperationTransformer.java | 5 +-
.../iceberg/operate/TestExpirationHandler.java | 143 +++++++++++
.../java/org/apache/drill/metastore/Metastore.java | 2 +-
.../apache/drill/metastore/MetastoreRegistry.java | 9 +-
.../main/resources/drill-metastore-default.conf | 2 +-
16 files changed, 536 insertions(+), 60 deletions(-)
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java
index 397e7bc..b3cd6f0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java
@@ -300,6 +300,7 @@ public class DrillbitContext implements AutoCloseable {
getFunctionImplementationRegistry().close();
getRemoteFunctionRegistry().close();
getCompiler().close();
+ metastoreRegistry.close();
}
public ResourceManager getResourceManager() {
diff --git a/metastore/iceberg-metastore/README.md b/metastore/iceberg-metastore/README.md
index 7d805c1..cc46c2a 100644
--- a/metastore/iceberg-metastore/README.md
+++ b/metastore/iceberg-metastore/README.md
@@ -15,6 +15,10 @@ can be overwritten in `drill-metastore-distrib.conf` or `drill-metastore-overri
`drill.metastore.iceberg.location.base_path` and `drill.metastore.iceberg.location.relative_path` -
indicate Iceberg Metastore base location.
+`drill.metastore.iceberg.expiration.period` - allows to specify period after which Iceberg table outdated
+metadata will be expired. Unit names must correspond to `java.time.temporal.ChronoUnit` enum values
+that do not have estimated duration (millis, seconds, minutes, hours, days).
+
`drill.metastore.iceberg.components` - provides configuration for specific Metastore components:
`drill.metastore.iceberg.components.tables`, `drill.metastore.iceberg.components.views`.
@@ -162,7 +166,7 @@ To delete data from Iceberg table, caller provides filter by which data will be
Filter expression must be based on component partition keys.
Delete operation removes partitions from Iceberg table, it does not remove data files to which
-these partitions were pointing.
+these partitions were pointing. Outdated data files will be deleted during expiration process.
If delete operation was successful, Iceberg table generates new snapshot and updates
its own metadata.
@@ -170,9 +174,16 @@ its own metadata.
### Purge
Allows to delete all data from Iceberg table. During this operation Iceberg table
-is not deleted, history of all operations and data files are preserved.
-
-## Data cleanup
-
-Iceberg table provides ability to remove outdated data files and snapshots
-when they are no longer needed. Such support in Iceberg Metastore will be added later.
+is not deleted, history of all operations and data files are preserved until
+expiration process is launched.
+
+## Iceberg metadata expiration
+
+Iceberg table generates metadata for each modification operation:
+snapshot, manifest file, table metadata file. Also when performing delete operation,
+previously stored data files are not deleted. These files with the time
+can occupy lots of space. `ExpirationHandler` allows to expire outdated metadata and
+data files after configured time period (`drill.metastore.iceberg.expiration.period`).
+If expiration period is not indicated, zero or negative, expiration won't be performed.
+`ExpirationHandler` is called after each modification operation, it checks if expiration period
+has elapsed and submits expiration process in a separate thread.
diff --git a/metastore/iceberg-metastore/pom.xml b/metastore/iceberg-metastore/pom.xml
index 1caaa7c..f8544fd 100644
--- a/metastore/iceberg-metastore/pom.xml
+++ b/metastore/iceberg-metastore/pom.xml
@@ -33,7 +33,7 @@
<name>metastore/Drill Iceberg Metastore</name>
<properties>
- <iceberg.version>cfd2737</iceberg.version>
+ <iceberg.version>77a456a</iceberg.version>
<caffeine.version>2.7.0</caffeine.version>
</properties>
diff --git a/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/IcebergMetastore.java b/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/IcebergMetastore.java
index 6cd375c..e223c5a 100644
--- a/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/IcebergMetastore.java
+++ b/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/IcebergMetastore.java
@@ -25,6 +25,7 @@ import org.apache.drill.metastore.components.views.Views;
import org.apache.drill.metastore.iceberg.components.tables.IcebergTables;
import org.apache.drill.metastore.iceberg.config.IcebergConfigConstants;
import org.apache.drill.metastore.iceberg.exceptions.IcebergMetastoreException;
+import org.apache.drill.metastore.iceberg.operate.ExpirationHandler;
import org.apache.drill.metastore.iceberg.schema.IcebergTableSchema;
import org.apache.drill.shaded.guava.com.google.common.collect.MapDifference;
import org.apache.drill.shaded.guava.com.google.common.collect.Maps;
@@ -58,6 +59,7 @@ public class IcebergMetastore implements Metastore {
private final org.apache.iceberg.Tables tables;
private final String baseLocation;
private final Map<String, String> commonProperties;
+ private final ExpirationHandler expirationHandler;
/**
* Table properties for each Iceberg table should be updated only once,
@@ -71,10 +73,11 @@ public class IcebergMetastore implements Metastore {
public IcebergMetastore(DrillConfig config) {
this.config = config;
- Configuration configuration = configuration(config);
+ Configuration configuration = configuration();
this.tables = new HadoopTables(new Configuration(configuration));
- this.baseLocation = baseLocation(config, new Configuration(configuration));
- this.commonProperties = properties(config, IcebergConfigConstants.COMPONENTS_COMMON_PROPERTIES);
+ this.baseLocation = baseLocation(new Configuration(configuration));
+ this.commonProperties = properties(IcebergConfigConstants.COMPONENTS_COMMON_PROPERTIES);
+ this.expirationHandler = new ExpirationHandler(config, new Configuration(configuration));
}
@Override
@@ -82,7 +85,7 @@ public class IcebergMetastore implements Metastore {
Table table = loadTable(IcebergConfigConstants.COMPONENTS_TABLES_LOCATION,
IcebergConfigConstants.COMPONENTS_TABLES_PROPERTIES,
IcebergTables.SCHEMA, Tables.class);
- return new IcebergTables(table);
+ return new IcebergTables(table, expirationHandler);
}
@Override
@@ -94,10 +97,9 @@ public class IcebergMetastore implements Metastore {
* Initializes {@link Configuration} based on config properties.
* if config properties are not indicated, returns default instance.
*
- * @param config Drill config
* @return {@link Configuration} instance
*/
- private Configuration configuration(DrillConfig config) {
+ private Configuration configuration() {
Configuration configuration = new Configuration();
if (config.hasPath(IcebergConfigConstants.CONFIG_PROPERTIES)) {
Config configProperties = config.getConfig(IcebergConfigConstants.CONFIG_PROPERTIES);
@@ -113,13 +115,12 @@ public class IcebergMetastore implements Metastore {
* If {@link IcebergConfigConstants#BASE_PATH} is not set, user home directory is used.
* {@link IcebergConfigConstants#RELATIVE_PATH} must be set.
*
- * @param config Drill config
* @param configuration Hadoop configuration
* @return Iceberg table base location
* @throws IcebergMetastoreException if unable to init file system
* or Iceberg Metastore relative path is not indicated
*/
- private String baseLocation(DrillConfig config, Configuration configuration) {
+ private String baseLocation(Configuration configuration) {
FileSystem fs;
try {
fs = FileSystem.get(configuration);
@@ -145,14 +146,13 @@ public class IcebergMetastore implements Metastore {
}
/**
- * Collects properties name and values into map if they are present in the given config,
+ * Collects properties name and values into map if they are present in the config,
* returns empty map otherwise.
*
- * @param config Drill config
* @param propertiesPath path to properties in the config
* @return map with properties names and their values
*/
- private Map<String, String> properties(DrillConfig config, String propertiesPath) {
+ private Map<String, String> properties(String propertiesPath) {
return config.hasPath(propertiesPath)
? config.getConfig(propertiesPath).entrySet().stream()
.collect(Collectors.toMap(
@@ -230,7 +230,7 @@ public class IcebergMetastore implements Metastore {
*/
private Map<String, String> tableProperties(String componentPropertiesConfig) {
Map<String, String> properties = new HashMap<>(commonProperties);
- properties.putAll(properties(config, componentPropertiesConfig));
+ properties.putAll(properties(componentPropertiesConfig));
return properties;
}
@@ -270,4 +270,9 @@ public class IcebergMetastore implements Metastore {
updateProperties.commit();
}
+
+ @Override
+ public void close() {
+ expirationHandler.close();
+ }
}
diff --git a/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/MetastoreContext.java b/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/MetastoreContext.java
index c4806d8..933910d 100644
--- a/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/MetastoreContext.java
+++ b/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/MetastoreContext.java
@@ -17,6 +17,7 @@
*/
package org.apache.drill.metastore.iceberg;
+import org.apache.drill.metastore.iceberg.operate.ExpirationHandler;
import org.apache.drill.metastore.iceberg.transform.Transformer;
import org.apache.drill.metastore.iceberg.write.FileWriter;
import org.apache.iceberg.Table;
@@ -50,4 +51,11 @@ public interface MetastoreContext<T> {
* @return transformer instance
*/
Transformer<T> transformer();
+
+ /**
+ * Returns expiration handler that expires Iceberg table metadata.
+ *
+ * @return expiration handler instance
+ */
+ ExpirationHandler expirationHandler();
}
diff --git a/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/components/tables/IcebergTables.java b/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/components/tables/IcebergTables.java
index c28c0ec..f4d32e9 100644
--- a/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/components/tables/IcebergTables.java
+++ b/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/components/tables/IcebergTables.java
@@ -18,6 +18,7 @@
package org.apache.drill.metastore.iceberg.components.tables;
import org.apache.drill.metastore.components.tables.Tables;
+import org.apache.drill.metastore.iceberg.operate.ExpirationHandler;
import org.apache.drill.metastore.operate.Metadata;
import org.apache.drill.metastore.operate.Modify;
import org.apache.drill.metastore.operate.Read;
@@ -55,9 +56,11 @@ public class IcebergTables implements Tables, MetastoreContext<TableMetadataUnit
public static IcebergTableSchema SCHEMA = IcebergTableSchema.of(TableMetadataUnit.class, PARTITION_KEYS);
private final Table table;
+ private final ExpirationHandler expirationHandler;
- public IcebergTables(Table table) {
+ public IcebergTables(Table table, ExpirationHandler expirationHandler) {
this.table = table;
+ this.expirationHandler = expirationHandler;
}
public MetastoreContext<TableMetadataUnit> context() {
@@ -93,4 +96,9 @@ public class IcebergTables implements Tables, MetastoreContext<TableMetadataUnit
public Transformer<TableMetadataUnit> transformer() {
return new TablesTransformer(context());
}
+
+ @Override
+ public ExpirationHandler expirationHandler() {
+ return expirationHandler;
+ }
}
diff --git a/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/config/IcebergConfigConstants.java b/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/config/IcebergConfigConstants.java
index 6f326ac..4419f68 100644
--- a/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/config/IcebergConfigConstants.java
+++ b/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/config/IcebergConfigConstants.java
@@ -52,6 +52,11 @@ public interface IcebergConfigConstants {
String RELATIVE_PATH = LOCATION_NAMESPACE + "relative_path";
/**
+ * Defines config which provides expiration period value.
+ */
+ String EXPIRATION_PERIOD = BASE + "expiration.period";
+
+ /**
* Drill Iceberg Metastore components configuration properties namespace.
*/
String COMPONENTS = BASE + "components.";
diff --git a/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/operate/ExpirationHandler.java b/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/operate/ExpirationHandler.java
new file mode 100644
index 0000000..f06d496
--- /dev/null
+++ b/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/operate/ExpirationHandler.java
@@ -0,0 +1,276 @@
+/*
+ * 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.metastore.iceberg.operate;
+
+import com.typesafe.config.ConfigValue;
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.metastore.iceberg.config.IcebergConfigConstants;
+import org.apache.drill.metastore.iceberg.exceptions.IcebergMetastoreException;
+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 org.apache.iceberg.Table;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Iceberg table generates metadata for each modification operation:
+ * snapshot, manifest file, table metadata file. Also when performing delete operation,
+ * previously stored data files are not deleted. These files with the time
+ * can occupy lots of space.
+ * <p/>
+ * Expiration handler expires outdated metadata and data files after configured expiration period.
+ * Expiration period is set in the Iceberg Metastore config {@link IcebergConfigConstants#EXPIRATION_PERIOD}.
+ * Units should correspond to {@link ChronoUnit} values that do not have estimated duration
+ * (millis, seconds, minutes, hours, days).
+ * If expiration period is not set, zero or negative, expiration process will not be executed.
+ * <p/>
+ * Expiration process is launched using executor service which allows to execute only one thread at a time,
+ * idle thread is not kept in the core pool since it is assumed that expiration process won't be launched to often.
+ * <p/>
+ * During Drillbit shutdown, if there are expiration tasks in the queue, they will be discarded in order to
+ * unblock Drillbit shutdown process.
+ */
+public class ExpirationHandler implements AutoCloseable {
+
+ private static final Logger logger = LoggerFactory.getLogger(ExpirationHandler.class);
+
+ private static final Pattern METADATA_VERSION_PATTERN = Pattern.compile("^v([0-9]+)\\..*");
+
+ // contains Iceberg table location and its last expiration time
+ private final Map<String, Long> expirationStatus = new ConcurrentHashMap<>();
+ private final Configuration configuration;
+ private final long expirationPeriod;
+ private volatile ExecutorService executorService;
+
+ public ExpirationHandler(DrillConfig config, Configuration configuration) {
+ this.configuration = configuration;
+ this.expirationPeriod = expirationPeriod(config);
+ logger.debug("Drill Iceberg Metastore expiration period: {}", expirationPeriod);
+ }
+
+ /**
+ * Checks if expiration process needs to be performed for the given Iceberg table
+ * by comparing stored last expiration time.
+ * If difference between last expiration time and current time is more or equal to
+ * expiration period, launches expiration process.
+ * If expiration period is zero or negative, no expiration process will be launched.
+ *
+ * @param table Iceberg table instance
+ * @return true if expiration process was launched, false otherwise
+ */
+ public boolean expire(Table table) {
+ if (expirationPeriod <= 0) {
+ return false;
+ }
+
+ long current = System.currentTimeMillis();
+ Long last = expirationStatus.putIfAbsent(table.location(), current);
+
+ if (last != null && current - last >= expirationPeriod) {
+ expirationStatus.put(table.location(), current);
+
+ ExecutorService executorService = executorService();
+ executorService.submit(() -> {
+ logger.debug("Expiring Iceberg table [{}] metadata", table.location());
+ table.expireSnapshots()
+ .expireOlderThan(current)
+ .commit();
+ // TODO: Replace with table metadata expiration through Iceberg API
+ // when https://github.com/apache/incubator-iceberg/issues/181 is resolved
+ // table.expireTableMetadata().expireOlderThan(current).commit();
+ expireTableMetadata(table);
+ });
+ return true;
+ }
+ return false;
+ }
+
+ public long expirationPeriod() {
+ return expirationPeriod;
+ }
+
+ @Override
+ public void close() {
+ if (executorService != null) {
+ // unlike shutdown(), shutdownNow() discards all queued waiting tasks
+ // this is done in order to unblock Drillbit shutdown
+ executorService.shutdownNow();
+ }
+ }
+
+ private long expirationPeriod(DrillConfig config) {
+ if (config.hasPath(IcebergConfigConstants.EXPIRATION_PERIOD)) {
+ Duration duration = config.getConfig(IcebergConfigConstants.EXPIRATION_PERIOD).entrySet().stream()
+ .map(this::duration)
+ .reduce(Duration.ZERO, Duration::plus);
+ return duration.toMillis();
+ }
+ return 0;
+ }
+
+ private Duration duration(Map.Entry<String, ConfigValue> entry) {
+ String amountText = String.valueOf(entry.getValue().unwrapped());
+ String unitText = entry.getKey().toUpperCase();
+ try {
+ long amount = Long.parseLong(amountText);
+ ChronoUnit unit = ChronoUnit.valueOf(unitText);
+ return Duration.of(amount, unit);
+ } catch (NumberFormatException e) {
+ throw new IcebergMetastoreException(String.format("Error when parsing expiration period config. " +
+ "Unable to convert [%s] into long", amountText), e);
+ } catch (IllegalArgumentException e) {
+ throw new IcebergMetastoreException(String.format("Error when parsing expiration period config. " +
+ "Unable to convert [%s] into [%s]", unitText, ChronoUnit.class.getCanonicalName()), e);
+ }
+ }
+
+ /**
+ * Initializes executor service instance using DCL.
+ * Created thread executor instance allows to execute only one thread at a time
+ * but unlike single thread executor does not keep this thread in the pool.
+ * Custom thread factory is used to define Iceberg Metastore specific thread names.
+ *
+ * @return executor service instance
+ */
+ private ExecutorService executorService() {
+ if (executorService == null) {
+ synchronized (this) {
+ if (executorService == null) {
+ this.executorService = new ThreadPoolExecutor(0, 1, 0L,
+ TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new IcebergThreadFactory());
+ }
+ }
+ }
+ return executorService;
+ }
+
+ /**
+ * Expires outdated Iceberg table metadata files.
+ * Reads current Iceberg table metadata version from version-hint.text file
+ * and deletes all metadata files that end with ".metadata.json" and have
+ * version less than current.
+ * <p/>
+ * Should be replaced with
+ * <code>table.expireTableMetadata().expireOlderThan(current).commit();</code>
+ * when <a href="https://github.com/apache/incubator-iceberg/issues/181">Issue#181</a>
+ * is resolved.
+ *
+ * @param table Iceberg table instance
+ */
+ private void expireTableMetadata(Table table) {
+ try {
+ String location = table.location();
+ Path metadata = new Path(location, "metadata");
+ FileSystem fs = metadata.getFileSystem(configuration);
+ for (FileStatus fileStatus : listExpiredMetadataFiles(fs, metadata)) {
+ if (fs.delete(fileStatus.getPath(), false)) {
+ logger.debug("Deleted Iceberg table [{}] metadata file [{}]", table.location(), fileStatus.getPath());
+ }
+ }
+ } catch (NumberFormatException | IOException e) {
+ logger.warn("Unable to expire Iceberg table [{}] metadata files", table.location(), e);
+ }
+ }
+
+ /**
+ * Reads current Iceberg table metadata version from version-hint.text file
+ * and returns all metadata files that end with ".metadata.json" and have
+ * version less than current.
+ *
+ * @param fs file system
+ * @param metadata pth to Iceberg metadata
+ * @return metadata files with version less than current
+ * @throws IOException in case of error listing file statuses
+ */
+ private FileStatus[] listExpiredMetadataFiles(FileSystem fs, Path metadata) throws IOException {
+ int currentVersion = currentVersion(fs, metadata);
+ return fs.listStatus(metadata, path -> {
+ if (path.getName().endsWith(".metadata.json")) {
+ int version = parseVersion(path);
+ return version != -1 && currentVersion > version;
+ }
+ return false;
+ });
+ }
+
+ /**
+ * Reads current table metadata version from version-hint.text file.
+ *
+ * @param fs file system
+ * @param metadata table metadata path
+ * @return current table metadata version
+ * @throws IOException if unable to read current table metadata version
+ */
+ private int currentVersion(FileSystem fs, Path metadata) throws IOException {
+ Path versionHintFile = new Path(metadata, "version-hint.text");
+ try (BufferedReader in = new BufferedReader(new InputStreamReader(fs.open(versionHintFile)))) {
+ return Integer.parseInt(in.readLine().replace("\n", ""));
+ }
+ }
+
+ /**
+ * Extracts metadata version from table metadata file name.
+ * Example: v1.metadata.json -> 1, v15.metadata.json -> 15
+ *
+ * @param path table metadata file path
+ * @return metadata version
+ */
+ private int parseVersion(Path path) {
+ Matcher matcher = METADATA_VERSION_PATTERN.matcher(path.getName());
+ if (matcher.find() && matcher.groupCount() == 1) {
+ return Integer.parseInt(matcher.group(1));
+ }
+ throw new NumberFormatException("Unable to parse version for path " + path);
+ }
+
+ /**
+ * Wraps default thread factory and adds Iceberg Metastore prefix to the original thread name.
+ * Is used to uniquely identify Iceberg metastore threads.
+ * Example: drill-iceberg-metastore-pool-1-thread-1
+ */
+ private static class IcebergThreadFactory implements ThreadFactory {
+
+ private static final String THREAD_PREFIX = "drill-iceberg-metastore-";
+ private final ThreadFactory delegate = Executors.defaultThreadFactory();
+
+ @Override
+ public Thread newThread(Runnable runnable) {
+ Thread thread = delegate.newThread(runnable);
+ thread.setName(THREAD_PREFIX + thread.getName());
+ return thread;
+ }
+ }
+}
diff --git a/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/operate/IcebergModify.java b/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/operate/IcebergModify.java
index a05c2dc..a510f86 100644
--- a/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/operate/IcebergModify.java
+++ b/metastore/iceberg-metastore/src/main/java/org/apache/drill/metastore/iceberg/operate/IcebergModify.java
@@ -79,5 +79,8 @@ public class IcebergModify<T> implements Modify<T> {
Transaction transaction = context.table().newTransaction();
operations.forEach(op -> op.add(transaction));
transaction.commitTransaction();
+
+ // check if Iceberg table metadata needs to be expired after each modification operation
+ context.expirationHandler().expire(context.table());
}
}
diff --git a/metastore/iceberg-metastore/src/main/resources/drill-metastore-module.conf b/metastore/iceberg-metastore/src/main/resources/drill-metastore-module.conf
index 182d029..be98a97 100644
--- a/metastore/iceberg-metastore/src/main/resources/drill-metastore-module.conf
+++ b/metastore/iceberg-metastore/src/main/resources/drill-metastore-module.conf
@@ -16,45 +16,53 @@
drill.metastore.iceberg: {
- // File system config can be specified
- config.properties: {
- // Iceberg tables location will be created under default file system
- // taken from Hadoop configuration unless the below property is set
- // fs.defaultFS: "file:///"
- }
+ // File system config can be specified
+ config.properties: {
+ // Iceberg tables location will be created under default file system
+ // taken from Hadoop configuration unless the below property is set
+ // fs.defaultFS: "file:///"
+ }
- // Iceberg Metastore location is constructed based on
- // combination of base_path and relative_path config values
- location: {
- // Iceberg table base path, if not indicated, user home directory will be used
- // base_path: "/app",
+ // Iceberg Metastore location is constructed based on
+ // combination of base_path and relative_path config values
+ location: {
+ // Iceberg table base path, if not indicated, user home directory will be used
+ // base_path: "/app",
- // Iceberg table relative path unique among clusters
- // ${drill.exec.zk.root} value will be substituted from Drill main config
- relative_path: ${drill.exec.zk.root}"/metastore/iceberg"
- }
+ // Iceberg table relative path unique among clusters
+ // ${drill.exec.zk.root} value will be substituted from Drill main config
+ relative_path: ${drill.exec.zk.root}"/metastore/iceberg"
+ }
+
+ // Specifies time period after which outdated Iceberg table metadata will be expired,
+ // unit names must correspond to java.time.temporal.ChronoUnit enum values
+ // that do not have estimated duration (millis, seconds, minutes, hours, days).
+ // Example: hours: 10, minutes: 20
+ expiration.period: {
+ days: 5
+ }
- components: {
- // Common properties for all Iceberg tables from org.apache.iceberg.TableProperties can be specified
- common.properties: {
- write.format.default: "parquet",
- write.metadata.compression-codec: "none"
- },
+ components: {
+ // Common properties for all Iceberg tables from org.apache.iceberg.TableProperties can be specified
+ common.properties: {
+ write.format.default: "parquet",
+ write.metadata.compression-codec: "none"
+ },
- tables: {
- // Iceberg table location in Iceberg Metastore
- location: "tables",
+ tables: {
+ // Iceberg table location in Iceberg Metastore
+ location: "tables",
- // Specific tables properties from org.apache.iceberg.TableProperties can be specified
- properties: {}
- },
+ // Specific tables properties from org.apache.iceberg.TableProperties can be specified
+ properties: {}
+ },
- views: {
- // Iceberg table location in Iceberg Metastore
- location: "views",
+ views: {
+ // Iceberg table location in Iceberg Metastore
+ location: "views",
- // Specific views properties from org.apache.iceberg.TableProperties can be specified
- properties: {}
- }
+ // Specific views properties from org.apache.iceberg.TableProperties can be specified
+ properties: {}
}
+ }
}
diff --git a/metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/components/tables/TestTableKey.java b/metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/components/tables/TestTableKey.java
index c84a3b9..3e96ca2 100644
--- a/metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/components/tables/TestTableKey.java
+++ b/metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/components/tables/TestTableKey.java
@@ -48,7 +48,7 @@ public class TestTableKey extends IcebergBaseTest {
TableKey tableKey = new TableKey("dfs", "tmp", "nation");
String expected = new Path(
- Paths.get("/metastore", "dfs", "tmp", "nation").toUri().getPath())
+ Paths.get("/metastore", "dfs", "tmp", "nation").toFile().getPath())
.toUri().getPath();
assertEquals(expected, tableKey.toLocation("/metastore"));
diff --git a/metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/components/tables/TestTablesOperationTransformer.java b/metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/components/tables/TestTablesOperationTransformer.java
index d7b72eb..80c6f77 100644
--- a/metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/components/tables/TestTablesOperationTransformer.java
+++ b/metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/components/tables/TestTablesOperationTransformer.java
@@ -70,9 +70,10 @@ public class TestTablesOperationTransformer extends IcebergBaseTest {
assertEquals(expression.toString(), operation.filter().toString());
- File file = new File(new Path(String.valueOf(operation.dataFile().path())).toUri().getPath());
+ Path path = new Path(String.valueOf(operation.dataFile().path()));
+ File file = new File(path.toUri().getPath());
assertTrue(file.exists());
- assertEquals(location, file.getParent());
+ assertEquals(location, path.getParent().toUri().getPath());
}
@Test
diff --git a/metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/operate/TestExpirationHandler.java b/metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/operate/TestExpirationHandler.java
new file mode 100644
index 0000000..90ef55b
--- /dev/null
+++ b/metastore/iceberg-metastore/src/test/java/org/apache/drill/metastore/iceberg/operate/TestExpirationHandler.java
@@ -0,0 +1,143 @@
+/*
+ * 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.metastore.iceberg.operate;
+
+import com.typesafe.config.ConfigValueFactory;
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.metastore.iceberg.IcebergBaseTest;
+import org.apache.drill.metastore.iceberg.config.IcebergConfigConstants;
+import org.apache.drill.metastore.iceberg.exceptions.IcebergMetastoreException;
+import org.apache.iceberg.Table;
+import org.junit.Test;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TestExpirationHandler extends IcebergBaseTest {
+
+ @Test
+ public void testConfigEmpty() {
+ ExpirationHandler expirationHandler = new ExpirationHandler(DrillConfig.create(), baseHadoopConfig());
+ assertEquals(0, expirationHandler.expirationPeriod());
+ }
+
+ @Test
+ public void testConfigOneUnit() {
+ DrillConfig config = new DrillConfig(DrillConfig.create()
+ .withValue(IcebergConfigConstants.EXPIRATION_PERIOD + ".hours",
+ ConfigValueFactory.fromAnyRef(5)));
+ ExpirationHandler expirationHandler = new ExpirationHandler(config, baseHadoopConfig());
+ assertEquals(TimeUnit.HOURS.toMillis(5), expirationHandler.expirationPeriod());
+ }
+
+ @Test
+ public void testConfigSeveralUnits() {
+ DrillConfig config = new DrillConfig(DrillConfig.create()
+ .withValue(IcebergConfigConstants.EXPIRATION_PERIOD + ".hours",
+ ConfigValueFactory.fromAnyRef(5))
+ .withValue(IcebergConfigConstants.EXPIRATION_PERIOD + ".minutes",
+ ConfigValueFactory.fromAnyRef(10)));
+ ExpirationHandler expirationHandler = new ExpirationHandler(config, baseHadoopConfig());
+ assertEquals(TimeUnit.HOURS.toMillis(5) + TimeUnit.MINUTES.toMillis(10),
+ expirationHandler.expirationPeriod());
+ }
+
+ @Test
+ public void testConfigNegativeValue() {
+ DrillConfig config = new DrillConfig(DrillConfig.create()
+ .withValue(IcebergConfigConstants.EXPIRATION_PERIOD + ".hours",
+ ConfigValueFactory.fromAnyRef(-5)));
+ ExpirationHandler expirationHandler = new ExpirationHandler(config, baseHadoopConfig());
+ assertEquals(TimeUnit.HOURS.toMillis(-5), expirationHandler.expirationPeriod());
+ }
+
+ @Test
+ public void testConfigIncorrectUnit() {
+ DrillConfig config = new DrillConfig(DrillConfig.create()
+ .withValue(IcebergConfigConstants.EXPIRATION_PERIOD + ".hour",
+ ConfigValueFactory.fromAnyRef(5)));
+
+ thrown.expect(IcebergMetastoreException.class);
+ new ExpirationHandler(config, baseHadoopConfig());
+ }
+
+ @Test
+ public void testConfigIncorrectValue() {
+ DrillConfig config = new DrillConfig(DrillConfig.create()
+ .withValue(IcebergConfigConstants.EXPIRATION_PERIOD + ".hours",
+ ConfigValueFactory.fromAnyRef("abc")));
+
+ thrown.expect(IcebergMetastoreException.class);
+ new ExpirationHandler(config, baseHadoopConfig());
+ }
+
+ @Test
+ public void testExpireZeroExpirationPeriod() {
+ DrillConfig config = new DrillConfig(DrillConfig.create()
+ .withValue(IcebergConfigConstants.EXPIRATION_PERIOD + ".millis",
+ ConfigValueFactory.fromAnyRef(0)));
+
+ ExpirationHandler expirationHandler = new ExpirationHandler(config, baseHadoopConfig());
+ Table table = mock(Table.class);
+ assertFalse(expirationHandler.expire(table));
+ }
+
+ @Test
+ public void testExpireNegativeExpirationPeriod() {
+ DrillConfig config = new DrillConfig(DrillConfig.create()
+ .withValue(IcebergConfigConstants.EXPIRATION_PERIOD + ".millis",
+ ConfigValueFactory.fromAnyRef(-10)));
+
+ ExpirationHandler expirationHandler = new ExpirationHandler(config, baseHadoopConfig());
+ Table table = mock(Table.class);
+ assertFalse(expirationHandler.expire(table));
+ }
+
+ @Test
+ public void testExpireFirstTime() {
+ DrillConfig config = new DrillConfig(DrillConfig.create()
+ .withValue(IcebergConfigConstants.EXPIRATION_PERIOD + ".millis",
+ ConfigValueFactory.fromAnyRef(1)));
+
+ ExpirationHandler expirationHandler = new ExpirationHandler(config, baseHadoopConfig());
+
+ Table table = mock(Table.class);
+ when(table.location()).thenReturn("/tmp/table");
+
+ assertFalse(expirationHandler.expire(table));
+ }
+
+ @Test
+ public void testExpireBefore() {
+ DrillConfig config = new DrillConfig(DrillConfig.create()
+ .withValue(IcebergConfigConstants.EXPIRATION_PERIOD + ".days",
+ ConfigValueFactory.fromAnyRef(1)));
+
+ ExpirationHandler expirationHandler = new ExpirationHandler(config, baseHadoopConfig());
+
+ Table table = mock(Table.class);
+ when(table.location()).thenReturn("/tmp/table");
+
+ assertFalse(expirationHandler.expire(table));
+ assertFalse(expirationHandler.expire(table));
+ }
+}
diff --git a/metastore/metastore-api/src/main/java/org/apache/drill/metastore/Metastore.java b/metastore/metastore-api/src/main/java/org/apache/drill/metastore/Metastore.java
index 7dad82d..2bbe461 100644
--- a/metastore/metastore-api/src/main/java/org/apache/drill/metastore/Metastore.java
+++ b/metastore/metastore-api/src/main/java/org/apache/drill/metastore/Metastore.java
@@ -29,7 +29,7 @@ import org.apache.drill.metastore.components.views.Views;
* Besides implementing {@link Metastore}, Metastore implementation must have constructor
* which accepts {@link org.apache.drill.common.config.DrillConfig}.
*/
-public interface Metastore {
+public interface Metastore extends AutoCloseable {
/**
* @return Metastore Tables component implementation
diff --git a/metastore/metastore-api/src/main/java/org/apache/drill/metastore/MetastoreRegistry.java b/metastore/metastore-api/src/main/java/org/apache/drill/metastore/MetastoreRegistry.java
index 0343463..293ab56 100644
--- a/metastore/metastore-api/src/main/java/org/apache/drill/metastore/MetastoreRegistry.java
+++ b/metastore/metastore-api/src/main/java/org/apache/drill/metastore/MetastoreRegistry.java
@@ -32,7 +32,7 @@ import java.lang.invoke.MethodType;
* Metastore initialization is delayed until {@link #get()} method is called.
* Metastore implementation must have constructor which accepts {@link DrillConfig}.
*/
-public class MetastoreRegistry {
+public class MetastoreRegistry implements AutoCloseable {
private DrillConfig config;
private volatile Metastore metastore;
@@ -104,4 +104,11 @@ public class MetastoreRegistry {
private DrillConfig createMetastoreConfig(DrillConfig config) {
return DrillConfig.create(null, null, true, new MetastoreConfigFileInfo(), config.root());
}
+
+ @Override
+ public void close() throws Exception {
+ if (metastore != null) {
+ metastore.close();
+ }
+ }
}
diff --git a/metastore/metastore-api/src/main/resources/drill-metastore-default.conf b/metastore/metastore-api/src/main/resources/drill-metastore-default.conf
index beb4f0a..3ab55b1 100644
--- a/metastore/metastore-api/src/main/resources/drill-metastore-default.conf
+++ b/metastore/metastore-api/src/main/resources/drill-metastore-default.conf
@@ -15,5 +15,5 @@
// limitations under the License.
drill.metastore: {
- implementation.class: "org.apache.drill.metastore.iceberg.IcebergMetastore"
+ implementation.class: "org.apache.drill.metastore.iceberg.IcebergMetastore"
}