|
| 1 | +// Licensed to the Apache Software Foundation (ASF) under one |
| 2 | +// or more contributor license agreements. See the NOTICE file |
| 3 | +// distributed with this work for additional information |
| 4 | +// regarding copyright ownership. The ASF licenses this file |
| 5 | +// to you under the Apache License, Version 2.0 (the |
| 6 | +// "License"); you may not use this file except in compliance |
| 7 | +// with the License. You may obtain a copy of the License at |
| 8 | +// |
| 9 | +// http://www.apache.org/licenses/LICENSE-2.0 |
| 10 | +// |
| 11 | +// Unless required by applicable law or agreed to in writing, |
| 12 | +// software distributed under the License is distributed on an |
| 13 | +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 14 | +// KIND, either express or implied. See the License for the |
| 15 | +// specific language governing permissions and limitations |
| 16 | +// under the License. |
| 17 | + |
| 18 | +package org.apache.doris.connector.paimon; |
| 19 | + |
| 20 | +import org.apache.doris.connector.cache.ConnectorMetadataCache; |
| 21 | +import org.apache.doris.connector.spi.mvcc.ConnectorTimeTravelSpec; |
| 22 | + |
| 23 | +import org.apache.paimon.catalog.Identifier; |
| 24 | +import org.apache.paimon.types.DataTypes; |
| 25 | +import org.apache.paimon.types.RowType; |
| 26 | +import org.junit.jupiter.api.Assertions; |
| 27 | +import org.junit.jupiter.api.Test; |
| 28 | +import org.junit.jupiter.params.ParameterizedTest; |
| 29 | +import org.junit.jupiter.params.provider.MethodSource; |
| 30 | +import org.junit.jupiter.params.provider.ValueSource; |
| 31 | + |
| 32 | +import java.util.Collections; |
| 33 | +import java.util.Map; |
| 34 | +import java.util.OptionalLong; |
| 35 | +import java.util.stream.Stream; |
| 36 | + |
| 37 | +public class PaimonSnapshotAuthenticationTest { |
| 38 | + private final RecordingConnectorContext context = new RecordingConnectorContext(); |
| 39 | + private final RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps(); |
| 40 | + |
| 41 | + private PaimonTableHandle handle() { |
| 42 | + FakePaimonTable table = new FakePaimonTable("t", |
| 43 | + RowType.builder().field("region", DataTypes.STRING()).build(), |
| 44 | + Collections.singletonList("region"), Collections.emptyList()); |
| 45 | + ops.table = table; |
| 46 | + ops.latestSnapshotId = OptionalLong.of(7L); |
| 47 | + ops.snapshotReadAuthentication = () -> context.authDepth > 0; |
| 48 | + PaimonTableHandle handle = new PaimonTableHandle("db", "t", |
| 49 | + Collections.singletonList("region"), Collections.emptyList()); |
| 50 | + handle.setPaimonTable(table); |
| 51 | + return handle; |
| 52 | + } |
| 53 | + |
| 54 | + private PaimonConnectorMetadata metadata(PaimonLatestSnapshotCache cache) { |
| 55 | + return new PaimonConnectorMetadata(ops, PaimonCatalogProperties.of(Collections.emptyMap()), context, |
| 56 | + new PaimonSchemaAtMemo(10), cache, |
| 57 | + new ConnectorMetadataCache<>("paimon", "partition_view", Collections.emptyMap())); |
| 58 | + } |
| 59 | + |
| 60 | + @ParameterizedTest |
| 61 | + @ValueSource(longs = {0L, 60L}) |
| 62 | + public void snapshotCacheLoadsInsideAuthentication(long ttl) { |
| 63 | + PaimonTableHandle handle = handle(); |
| 64 | + PaimonLatestSnapshotCache cache = new PaimonLatestSnapshotCache(ttl, 10); |
| 65 | + Assertions.assertEquals(7L, metadata(cache).beginQuerySnapshot(null, handle).get().getSnapshotId()); |
| 66 | + ops.latestSnapshotId = OptionalLong.of(8L); |
| 67 | + Assertions.assertEquals(ttl == 0 ? 8L : 7L, |
| 68 | + metadata(cache).beginQuerySnapshot(null, handle).get().getSnapshotId()); |
| 69 | + cache.invalidate(Identifier.create("db", "t")); |
| 70 | + Assertions.assertEquals(8L, metadata(cache).beginQuerySnapshot(null, handle).get().getSnapshotId()); |
| 71 | + Assertions.assertEquals(0, context.authDepth); |
| 72 | + } |
| 73 | + |
| 74 | + @Test |
| 75 | + public void partitionCacheKeyLoadsSnapshotInsideAuthentication() { |
| 76 | + PaimonTableHandle handle = handle(); |
| 77 | + Assertions.assertTrue(metadata(new PaimonLatestSnapshotCache(0, 10)) |
| 78 | + .listPartitionNames(null, handle).isEmpty()); |
| 79 | + Assertions.assertTrue(ops.log.contains("latestSnapshotId")); |
| 80 | + Assertions.assertEquals(0, context.authDepth); |
| 81 | + } |
| 82 | + |
| 83 | + @Test |
| 84 | + public void emptySnapshotRemainsAnEmptyPin() { |
| 85 | + PaimonTableHandle handle = handle(); |
| 86 | + ops.latestSnapshotId = OptionalLong.empty(); |
| 87 | + Assertions.assertEquals(-1L, metadata(new PaimonLatestSnapshotCache(0, 10)) |
| 88 | + .beginQuerySnapshot(null, handle).get().getSnapshotId()); |
| 89 | + } |
| 90 | + |
| 91 | + @Test |
| 92 | + public void authenticationFailureDoesNotCacheAnEmptySnapshot() { |
| 93 | + PaimonTableHandle handle = handle(); |
| 94 | + PaimonLatestSnapshotCache cache = new PaimonLatestSnapshotCache(60, 10); |
| 95 | + context.failAuth = true; |
| 96 | + Assertions.assertThrows(RuntimeException.class, () -> metadata(cache).beginQuerySnapshot(null, handle)); |
| 97 | + Assertions.assertFalse(ops.log.contains("latestSnapshotId")); |
| 98 | + context.failAuth = false; |
| 99 | + Assertions.assertEquals(7L, metadata(cache).beginQuerySnapshot(null, handle).get().getSnapshotId()); |
| 100 | + Assertions.assertEquals(0, context.authDepth); |
| 101 | + } |
| 102 | + |
| 103 | + static Stream<ConnectorTimeTravelSpec> timeTravelSpecs() { |
| 104 | + return Stream.of(ConnectorTimeTravelSpec.snapshotId("7"), |
| 105 | + ConnectorTimeTravelSpec.timestamp("1700000000000", true), |
| 106 | + ConnectorTimeTravelSpec.tag("release"), |
| 107 | + ConnectorTimeTravelSpec.versionRef("release"), |
| 108 | + ConnectorTimeTravelSpec.branch("branch"), |
| 109 | + ConnectorTimeTravelSpec.incremental(Map.of("startSnapshotId", "1", "endSnapshotId", "7")), |
| 110 | + ConnectorTimeTravelSpec.options(Collections.emptyMap())); |
| 111 | + } |
| 112 | + |
| 113 | + @ParameterizedTest |
| 114 | + @MethodSource("timeTravelSpecs") |
| 115 | + public void timeTravelReadsInsideAuthentication(ConnectorTimeTravelSpec spec) { |
| 116 | + PaimonTableHandle handle = handle(); |
| 117 | + ops.snapshotExists = true; |
| 118 | + ops.branchExists = true; |
| 119 | + ops.branchTable = ops.table; |
| 120 | + ops.tagSnapshot = new PaimonCatalogOps.TagSnapshot(7L, 1L); |
| 121 | + ops.snapshotIdAtOrBefore = OptionalLong.of(7L); |
| 122 | + ops.snapshotSchemaId = OptionalLong.of(1L); |
| 123 | + Assertions.assertEquals(7L, metadata(new PaimonLatestSnapshotCache(0, 10)) |
| 124 | + .resolveTimeTravel(null, handle, spec).get().getSnapshotId()); |
| 125 | + Assertions.assertEquals(0, context.authDepth); |
| 126 | + } |
| 127 | +} |
0 commit comments