Skip to content

Allow accessing external environments in Delta Lake query runner #17202

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ebyhr merged 3 commits into from
Apr 27, 2023
Merged

Allow accessing external environments in Delta Lake query runner #17202

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2c26890
Remove unused method from DeltaLakeQueryRunner
ebyhr Apr 24, 2023
6396306
Move default file metastore from build method in Delta query runner
ebyhr Apr 24, 2023
9d37af9
Avoid hard-coded hive.metastore in DeltaLakeGlueQueryRunnerMain
ebyhr Apr 24, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 25 additions & 26 deletions plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/DeltaLakeQueryRunner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import io.trino.testing.QueryRunner;
import io.trino.tpch.TpchTable;

import java.nio.file.Files;
import java.nio.file.Path;
import java.util.HashMap;
import java.util.Map;
Expand Down Expand Up @@ -104,14 +105,7 @@ public DistributedQueryRunner build()
queryRunner.createCatalog("tpcds", "tpcds");

queryRunner.installPlugin(new TestingDeltaLakePlugin());
Map<String, String> deltaProperties = new HashMap<>(this.deltaProperties.buildOrThrow());
if (!deltaProperties.containsKey("hive.metastore") && !deltaProperties.containsKey("hive.metastore.uri")) {
Path dataDir = queryRunner.getCoordinator().getBaseDataDir().resolve(DELTA_CATALOG);
deltaProperties.put("hive.metastore", "file");
deltaProperties.put("hive.metastore.catalog.dir", dataDir.toUri().toString());
}

queryRunner.createCatalog(catalogName, CONNECTOR_NAME, deltaProperties);
queryRunner.createCatalog(catalogName, CONNECTOR_NAME, deltaProperties.buildOrThrow());

return queryRunner;
}
Expand All @@ -122,19 +116,21 @@ public DistributedQueryRunner build()
}
}

public static DistributedQueryRunner createDeltaLakeQueryRunner(String catalogName)
throws Exception
{
return createDeltaLakeQueryRunner(catalogName, ImmutableMap.of(), ImmutableMap.of());
}

public static DistributedQueryRunner createDeltaLakeQueryRunner(String catalogName, Map<String, String> extraProperties, Map<String, String> connectorProperties)
throws Exception
{
Map<String, String> deltaProperties = new HashMap<>(connectorProperties);
if (!deltaProperties.containsKey("hive.metastore") && !deltaProperties.containsKey("hive.metastore.uri")) {
Path metastoreDirectory = Files.createTempDirectory(catalogName);
metastoreDirectory.toFile().deleteOnExit();
deltaProperties.put("hive.metastore", "file");
deltaProperties.put("hive.metastore.catalog.dir", metastoreDirectory.toUri().toString());
}

DistributedQueryRunner queryRunner = builder(createSession())
.setCatalogName(catalogName)
.setExtraProperties(extraProperties)
.setDeltaProperties(connectorProperties)
.setDeltaProperties(deltaProperties)
.build();

queryRunner.execute("CREATE SCHEMA IF NOT EXISTS tpch");
Expand Down Expand Up @@ -247,14 +243,18 @@ public static class DefaultDeltaLakeQueryRunnerMain
public static void main(String[] args)
throws Exception
{
Path metastoreDirectory = Files.createTempDirectory(DELTA_CATALOG);
metastoreDirectory.toFile().deleteOnExit();
DistributedQueryRunner queryRunner = createDeltaLakeQueryRunner(
DELTA_CATALOG,
ImmutableMap.of("http-server.http.port", "8080"),
ImmutableMap.of("delta.enable-non-concurrent-writes", "true"));
ImmutableMap.of(
"delta.enable-non-concurrent-writes", "true",
"hive.metastore", "file",
"hive.metastore.catalog.dir", metastoreDirectory.toUri().toString()));

Path baseDirectory = queryRunner.getCoordinator().getBaseDataDir().resolve(DELTA_CATALOG);
copyTpchTables(queryRunner, "tpch", TINY_SCHEMA_NAME, createSession(), TpchTable.getTables());
log.info("Data directory is: %s", baseDirectory);
log.info("Data directory is: %s", metastoreDirectory);

Thread.sleep(10);
Logger log = Logger.get(DeltaLakeQueryRunner.class);
Expand All @@ -263,19 +263,18 @@ public static void main(String[] args)
}
}

public static class DeltaLakeGlueQueryRunnerMain
public static class DeltaLakeExternalQueryRunnerMain
{
public static void main(String[] args)
throws Exception
{
// Requires AWS credentials, which can be provided any way supported by the DefaultProviderChain
// See https://docs.aws.amazon.com/sdk-for-java/v1/developer-guide/credentials.html#credentials-default
DistributedQueryRunner queryRunner = createDeltaLakeQueryRunner(
DELTA_CATALOG,
ImmutableMap.of("http-server.http.port", "8080"),
ImmutableMap.of("hive.metastore", "glue"));
// Please set Delta Lake connector properties via VM options. e.g. -Dhive.metastore=glue -D..
DistributedQueryRunner queryRunner = builder()
.setCatalogName(DELTA_CATALOG)
.setExtraProperties(ImmutableMap.of("http-server.http.port", "8080"))
.build();

Logger log = Logger.get(DeltaLakeGlueQueryRunnerMain.class);
Logger log = Logger.get(DeltaLakeExternalQueryRunnerMain.class);
log.info("======== SERVER STARTED ========");
log.info("\n====\n%s\n====", queryRunner.getCoordinator().getBaseUrl());
}
Expand Down