findinpath
commented
2 years ago /*
* Licensed 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 io.trino.tests.product.deltalake;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.model.DeleteObjectsRequest;
import com.amazonaws.services.s3.model.ObjectListing;
import com.amazonaws.services.s3.model.S3ObjectSummary;
import com.google.inject.Inject;
import com.google.inject.name.Named;
import io.trino.tempto.BeforeTestWithContext;
import org.testng.annotations.Test;
import java.util.List;
import static io.trino.tempto.assertions.QueryAssert.Row.row;
import static io.trino.tempto.assertions.QueryAssert.assertThat;
import static io.trino.tests.product.TestGroups.DELTA_LAKE_DATABRICKS;
import static io.trino.tests.product.TestGroups.DELTA_LAKE_OSS;
import static io.trino.tests.product.TestGroups.PROFILE_SPECIFIC_TESTS;
import static io.trino.tests.product.hive.util.TemporaryHiveTable.randomTableSuffix;
import static io.trino.tests.product.utils.QueryExecutors.onDelta;
import static io.trino.tests.product.utils.QueryExecutors.onTrino;
import static java.lang.String.format;
public class TestDeltaLakeActiveFilesCache
extends BaseTestDeltaLakeS3Storage
{
@Inject
@Named("s3.server_type")
private String s3ServerType;
private AmazonS3 s3;
@BeforeTestWithContext
public void setup()
{
super.setUp();
s3 = new S3ClientFactory().createS3Client(s3ServerType);
}
@Test(groups = {DELTA_LAKE_DATABRICKS, DELTA_LAKE_OSS, PROFILE_SPECIFIC_TESTS})
public void testSelectFromTrinoShouldRefreshTheFilesCacheWhenTableIsRecreated()
{
String tableName = "test_dl_cached_table_files_refres_" + randomTableSuffix();
String tableDirectory = "databricks-compatibility-test-" + tableName;
onTrino().executeQuery(format("CREATE TABLE delta.default.%s (col INT) WITH (location = 's3://%s/%s')",
tableName,
bucketName,
tableDirectory));
onTrino().executeQuery("INSERT INTO " + tableName + " VALUES 1");
// Add the files of the table in the active files cache
assertThat(onTrino().executeQuery("SELECT * FROM " + tableName)).containsOnly(row(1));
// Recreate the table outside of Trino to avoid updating the Trino table active files cache
onDelta().executeQuery("DROP TABLE default." + tableName);
// Delete the contents of the table explicitly from storage (because it has been created as `EXTERNAL`)
deleteDirectory(tableDirectory);
onDelta().executeQuery(format("CREATE TABLE default.%s (col INTEGER) USING DELTA LOCATION 's3://%s/%s'",
tableName,
bucketName,
tableDirectory));
onDelta().executeQuery("INSERT INTO default." + tableName + " VALUES 2");
assertThat(onTrino().executeQuery("SELECT * FROM " + tableName)).containsOnly(row(2));
onTrino().executeQuery("DROP TABLE " + tableName);
}
private void deleteDirectory(String tableDirectory)
{
ObjectListing objectList = s3.listObjects(this.bucketName, tableDirectory);
List<S3ObjectSummary> objectSummeryList = objectList.getObjectSummaries();
String[] keysList = new String[objectSummeryList.size()];
int count = 0;
for (S3ObjectSummary summery : objectSummeryList) {
keysList[count++] = summery.getKey();
}
DeleteObjectsRequest deleteObjectsRequest = new DeleteObjectsRequest(bucketName).withKeys(keysList);
s3.deleteObjects(deleteObjectsRequest);
}
}
homar
findepi
Scenario:
// Create table either in Spark or Trino // Select contents of the table in Trino (fill active files cache) // Drop table via Spark (and remove, if table is external) the content of the table // Create table via Spark // Select contents of the table in Trino
Stacktrace:
Cause: the table's active files cache in https://github.com/trinodb/trino/blob/6310e5e1415fe0bc89444016fc516640c3a3fcbb/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/transactionlog/TransactionLogAccess.java is not invalidated in case that the table is dropped and recreated externally which will cause the
SELECTstatement to look for files that don't exist anymore.