-
Notifications
You must be signed in to change notification settings - Fork 285
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
use as of system time when listing tables to improve performance
- Loading branch information
Yingjian Wu
committed
Jan 4, 2024
1 parent
2c0f438
commit 4fdfd86
Showing
9 changed files
with
213 additions
and
63 deletions.
There are no files selected for viewing
82 changes: 82 additions & 0 deletions
82
...ava/com/netflix/metacat/connector/polaris/PolarisConnectorTableServiceFunctionalTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| package com.netflix.metacat.connector.polaris; | ||
|
|
||
| import com.google.common.collect.ImmutableMap; | ||
| import com.google.common.collect.ImmutableSet; | ||
| import com.netflix.metacat.common.QualifiedName; | ||
| import com.netflix.metacat.common.dto.Pageable; | ||
| import com.netflix.metacat.common.dto.Sort; | ||
| import com.netflix.metacat.common.dto.SortOrder; | ||
| import com.netflix.metacat.common.server.connectors.model.TableInfo; | ||
| import com.netflix.metacat.connector.polaris.configs.PolarisPersistenceConfig; | ||
| import org.junit.Assert; | ||
| import org.junit.jupiter.api.Test; | ||
| import org.junit.jupiter.api.extension.ExtendWith; | ||
| import org.springframework.boot.test.autoconfigure.orm.jpa.AutoConfigureDataJpa; | ||
| import org.springframework.boot.test.context.SpringBootTest; | ||
| import org.springframework.test.context.ActiveProfiles; | ||
| import org.springframework.test.context.junit.jupiter.SpringExtension; | ||
|
|
||
| import java.util.List; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| /** | ||
| * Test PolarisConnectorTableService in functional test. | ||
| * Some of the tests cannot be run in unit test as it uses h2 database, which does not support all | ||
| * functionalities in crdb so include those tests here. | ||
| */ | ||
| @ExtendWith(SpringExtension.class) | ||
| @SpringBootTest(classes = {PolarisPersistenceConfig.class}) | ||
| @ActiveProfiles(profiles = {"polaris_functional_test"}) | ||
| @AutoConfigureDataJpa | ||
| public class PolarisConnectorTableServiceFunctionalTest extends PolarisConnectorTableServiceTest { | ||
| /** | ||
| * Test table list. | ||
| */ | ||
| @Test | ||
| public void testList() { | ||
| final QualifiedName name1 = QualifiedName.ofTable(CATALOG_NAME, DB_NAME, "table1"); | ||
| final TableInfo tableInfo1 = TableInfo.builder() | ||
| .name(name1) | ||
| .metadata(ImmutableMap.of("table_type", "ICEBERG", "metadata_location", "loc1")) | ||
| .build(); | ||
| this.getPolarisTableService().create(this.getRequestContext(), tableInfo1); | ||
| final QualifiedName name2 = QualifiedName.ofTable(CATALOG_NAME, DB_NAME, "table2"); | ||
| final TableInfo tableInfo2 = TableInfo.builder() | ||
| .name(name2) | ||
| .metadata(ImmutableMap.of("table_type", "ICEBERG", "metadata_location", "loc2")) | ||
| .build(); | ||
| this.getPolarisTableService().create(this.getRequestContext(), tableInfo2); | ||
|
|
||
|
|
||
| final QualifiedName qualifiedName = QualifiedName.ofTable(CATALOG_NAME, DB_NAME, ""); | ||
|
|
||
| try { | ||
| // pause execution for 10000 milliseconds (10 seconds) | ||
| Thread.sleep(10000); | ||
| } catch (InterruptedException e) { | ||
| System.out.println("Sleep was interrupted"); | ||
| } | ||
|
|
||
| List<TableInfo> tables = this.getPolarisTableService().list( | ||
| this.getRequestContext(), DB_QUALIFIED_NAME, qualifiedName, new Sort(null, SortOrder.ASC), | ||
| new Pageable(2, 0)); | ||
| Assert.assertEquals(tables.size(), 2); | ||
| Assert.assertEquals(tables.stream().map(TableInfo::getName).collect(Collectors.toSet()), | ||
| ImmutableSet.of(name1, name2)); | ||
|
|
||
| // Create a 3rd table, but this time does not sleep so this table should not be included | ||
| final QualifiedName name3 = QualifiedName.ofTable(CATALOG_NAME, DB_NAME, "table3"); | ||
| final TableInfo tableInfo3 = TableInfo.builder() | ||
| .name(name3) | ||
| .metadata(ImmutableMap.of("table_type", "ICEBERG", "metadata_location", "loc2")) | ||
| .build(); | ||
| this.getPolarisTableService().create(this.getRequestContext(), tableInfo3); | ||
|
|
||
| tables = this.getPolarisTableService().list( | ||
| this.getRequestContext(), DB_QUALIFIED_NAME, qualifiedName, new Sort(null, SortOrder.ASC), | ||
| new Pageable(2, 0)); | ||
| Assert.assertEquals(tables.size(), 2); | ||
| Assert.assertEquals(tables.stream().map(TableInfo::getName).collect(Collectors.toSet()), | ||
| ImmutableSet.of(name1, name2)); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
18 changes: 18 additions & 0 deletions
18
.../java/com/netflix/metacat/connector/polaris/store/repos/PolarisTableCustomRepository.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| package com.netflix.metacat.connector.polaris.store.repos; | ||
|
|
||
| import com.netflix.metacat.connector.polaris.store.entities.PolarisTableEntity; | ||
| import java.util.List; | ||
|
|
||
| /** | ||
| * Custom JPA repository implementation for storing PolarisTableEntity. | ||
| */ | ||
| public interface PolarisTableCustomRepository { | ||
| /** | ||
| * Fetch table entities for given database using AS OF SYSTEM TIME follower_read_timestamp(). | ||
| * @param dbName database name | ||
| * @param tableNamePrefix table name prefix. can be empty. | ||
| * @return table entities in the database. | ||
| */ | ||
| List<PolarisTableEntity> findAllTablesByDbNameAndTablePrefix( | ||
| String dbName, String tableNamePrefix); | ||
| } |
100 changes: 100 additions & 0 deletions
100
...a/com/netflix/metacat/connector/polaris/store/repos/PolarisTableCustomRepositoryImpl.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,100 @@ | ||
| package com.netflix.metacat.connector.polaris.store.repos; | ||
|
|
||
| import javax.persistence.EntityManagerFactory; | ||
| import javax.persistence.EntityManager; | ||
| import javax.persistence.Query; | ||
| import javax.persistence.EntityTransaction; | ||
|
|
||
| import com.netflix.metacat.connector.polaris.store.entities.PolarisTableEntity; | ||
| import org.springframework.beans.factory.annotation.Autowired; | ||
| import org.springframework.data.domain.Slice; | ||
| import org.springframework.data.domain.SliceImpl; | ||
| import org.springframework.data.domain.Pageable; | ||
| import org.springframework.stereotype.Repository; | ||
| import org.springframework.data.domain.PageRequest; | ||
| import org.springframework.data.domain.Sort; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
|
|
||
| /** | ||
| * Implementation for Custom JPA repository implementation for storing PolarisTableEntity. | ||
| */ | ||
| @Repository | ||
| public class PolarisTableCustomRepositoryImpl implements PolarisTableCustomRepository { | ||
| @Autowired | ||
| private EntityManagerFactory entityManagerFactory; | ||
|
|
||
| private Slice<PolarisTableEntity> findAllTablesByDbNameAndTablePrefixForCurrentPage( | ||
| final String dbName, final String tableNamePrefix, final Pageable page, final EntityManager entityManager) { | ||
|
|
||
| final String sql = "SELECT t.* FROM TBLS t " | ||
| + "WHERE t.db_name = :dbName AND t.tbl_name LIKE :tableNamePrefix"; | ||
|
|
||
| final Query query = entityManager.createNativeQuery(sql, PolarisTableEntity.class); | ||
| query.setParameter("dbName", dbName); | ||
| query.setParameter("tableNamePrefix", tableNamePrefix + "%"); | ||
|
|
||
| query.setFirstResult(page.getPageNumber() * page.getPageSize()); | ||
| query.setMaxResults(page.getPageSize()); | ||
|
|
||
| final List<PolarisTableEntity> resultList = query.getResultList(); | ||
|
|
||
| // Check if there is a next page | ||
| final boolean hasNext = resultList.size() > page.getPageSize(); | ||
|
|
||
| // If there is a next page, remove the last item from the list | ||
| if (hasNext) { | ||
| resultList.remove(resultList.size() - 1); | ||
| } | ||
|
|
||
| return new SliceImpl<>(resultList, page, hasNext); | ||
| } | ||
|
|
||
|
|
||
| @SuppressWarnings("checkstyle:FinalParameters") | ||
| @Override | ||
| public List<PolarisTableEntity> findAllTablesByDbNameAndTablePrefix( | ||
| final String dbName, final String tableNamePrefix) { | ||
| final int pageFetchSize = 1000; | ||
| Pageable page = PageRequest.of(0, pageFetchSize, Sort.by("tbl_name").ascending()); | ||
|
|
||
| final EntityManager entityManager = entityManagerFactory.createEntityManager(); | ||
| final EntityTransaction transaction = entityManager.getTransaction(); | ||
|
|
||
| try { | ||
| transaction.begin(); | ||
|
|
||
| // Execute your SQL statement | ||
| entityManager.createNativeQuery("SET TRANSACTION AS OF SYSTEM TIME follower_read_timestamp()") | ||
| .executeUpdate(); | ||
|
|
||
| final List<PolarisTableEntity> retval = new ArrayList<>(); | ||
| final String tblPrefix = tableNamePrefix == null ? "" : tableNamePrefix; | ||
| Slice<PolarisTableEntity> tbls; | ||
| boolean hasNext; | ||
|
|
||
| do { | ||
| tbls = findAllTablesByDbNameAndTablePrefixForCurrentPage(dbName, tblPrefix, page, entityManager); | ||
| retval.addAll(tbls.toList()); | ||
| hasNext = tbls.hasNext(); | ||
| if (hasNext) { | ||
| page = tbls.nextPageable(); | ||
| } | ||
| } while (hasNext); | ||
|
|
||
| transaction.commit(); | ||
|
|
||
| return retval; | ||
| } catch (Exception e) { | ||
| // If there's an exception, roll back the transaction | ||
| if (transaction.isActive()) { | ||
| transaction.rollback(); | ||
| } | ||
| // Handle or rethrow the exception | ||
| throw e; | ||
| } finally { | ||
| entityManager.close(); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters