Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
31 changes: 6 additions & 25 deletions fe/fe-common/src/main/java/org/apache/doris/common/Config.java
Original file line number Diff line number Diff line change
Expand Up @@ -4237,33 +4237,14 @@ public void handle(Field field, String value) throws Exception {
public static long lance_catalog_arrow_memory_limit_bytes = 256L * 1024 * 1024;

@ConfField(mutable = true, masterOnly = true, varType = VariableAnnotation.EXPERIMENTAL, description = {
"是否启用 Lance 外表索引变更(CREATE/CREATE OR REPLACE/DROP INDEX)的 admission。默认关闭;"
+ "启用前需确认未决 job 配额均为正值。注意:在 dispatch(后续版本)与 FORCE_RELEASE(后续版本)就绪前"
+ "开启本开关会产生不可回收的 PENDING job 并冻结对应 catalog 的身份属性变更与 DROP CATALOG。",
"Enable admission of Lance index mutations (CREATE/CREATE OR REPLACE/DROP INDEX). "
+ "Disabled by default; unresolved-job quotas must be positive before enabling. "
+ "WARNING: enabling before dispatch and FORCE_RELEASE land in a later release creates "
+ "PENDING jobs that cannot be resolved and freezes catalog identity changes and DROP CATALOG."})
"是否启用 Lance 外表索引变更(CREATE/CREATE OR REPLACE/DROP INDEX)的权威校验路径。默认关闭;"
+ "在同步执行路径就绪前,开启本开关只会让通过校验的语句收到暂不支持的错误。",
"Enable the authoritative validation path of Lance index mutations "
+ "(CREATE/CREATE OR REPLACE/DROP INDEX). Disabled by default; until the synchronous "
+ "execution path lands, enabling it only makes validated statements fail with the "
+ "not-supported error."})
public static boolean enable_lance_index_mutation = false;

@ConfField(mutable = true, masterOnly = true,
callback = LanceIndexConfigValidator.PositiveLongConfigHandler.class,
description = {"单个 Lance 数据表(locator 身份)允许的最大未决索引 job 数。",
"Max unresolved Lance index jobs per table (locator identity)."})
public static long lance_index_job_max_unresolved_per_table = 8;

@ConfField(mutable = true, masterOnly = true,
callback = LanceIndexConfigValidator.PositiveLongConfigHandler.class,
description = {"单个 Lance catalog 允许的最大未决索引 job 数。",
"Max unresolved Lance index jobs per catalog."})
public static long lance_index_job_max_unresolved_per_catalog = 64;

@ConfField(mutable = true, masterOnly = true,
callback = LanceIndexConfigValidator.PositiveLongConfigHandler.class,
description = {"全部 catalog 合计允许的最大未决 Lance 索引 job 数。",
"Max unresolved Lance index jobs across all catalogs (global)."})
public static long lance_index_job_max_unresolved_global = 256;

@ConfField(mutable = true, masterOnly = true,
callback = LanceIndexConfigValidator.PositiveIntConfigHandler.class,
description = {"Lance IVF_PQ 索引 num_partitions 的静态上限。",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,40 +20,19 @@
import java.lang.reflect.Field;

/**
* Validators for the Lance index admission configuration items in {@link Config}. The
* unresolved-job quotas and the static bounds must stay positive (section 9.7 enabling
* precondition), so every handler rejects a zero, negative, or unparsable value before
* assigning. The handlers extend {@link ConfigBase.DefaultConfHandler} and delegate to
* Validators for the Lance index configuration items in {@link Config}. The static bounds
* must stay positive, so the handler rejects a zero, negative, or unparsable value before
* assigning. The handler extends {@link ConfigBase.DefaultConfHandler} and delegates to
* {@code super.handle} after validation: a bare {@code ConfHandler} implementation would
* validate without ever assigning, letting ADMIN SET FRONTEND CONFIG pass with no effect.
* Note the callback only runs on the ADMIN SET path; values loaded from fe.conf bypass it
* ({@code ConfigBase.setFields} assigns directly), so the admission path re-asserts the
* positive invariant where the values are consumed.
* ({@code ConfigBase.setFields} assigns directly), so the consuming validation path
* re-asserts the positive invariant where the values are used.
*/
public final class LanceIndexConfigValidator {
private LanceIndexConfigValidator() {
}

/**
* Accepts only a positive long: validates first, then assigns via the default handler.
*/
public static class PositiveLongConfigHandler extends ConfigBase.DefaultConfHandler {
@Override
public void handle(Field field, String confVal) throws Exception {
String trimmedVal = confVal == null ? "" : confVal.trim();
final long value;
try {
value = Long.parseLong(trimmedVal);
} catch (NumberFormatException e) {
throw new ConfigException(field.getName() + " must be a positive long, but got: " + confVal, e);
}
if (value <= 0) {
throw new ConfigException(field.getName() + " must be a positive long, but got: " + confVal);
}
super.handle(field, trimmedVal);
}
}

/**
* Accepts only a positive int: validates first, then assigns via the default handler.
*/
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,8 @@
import java.nio.file.Path;

/**
* Coverage for the Lance index admission configuration items in {@link Config} and the
* positive-value validators in {@link LanceIndexConfigValidator}: reviewed defaults, the
* Coverage for the Lance index configuration items in {@link Config} and the positive-value
* validator in {@link LanceIndexConfigValidator}: reviewed defaults, the
* {@link ConfigBase.ConfField} wiring (mutable/masterOnly/callback/varType), rejection of
* zero, negative and non-numeric values, and that an accepted value is really assigned to
* the field (a bare validating handler that never assigns would let ADMIN SET pass without
Expand All @@ -37,21 +37,12 @@ public class LanceIndexConfigValidatorTest {
@Test
public void testReviewedDefaults() {
Assertions.assertFalse(Config.enable_lance_index_mutation);
Assertions.assertEquals(8L, Config.lance_index_job_max_unresolved_per_table);
Assertions.assertEquals(64L, Config.lance_index_job_max_unresolved_per_catalog);
Assertions.assertEquals(256L, Config.lance_index_job_max_unresolved_global);
Assertions.assertEquals(4096, Config.lance_index_max_num_partitions);
Assertions.assertEquals(256, Config.lance_index_max_num_sub_vectors);
}

@Test
public void testConfFieldWiring() throws Exception {
assertCallbackWiring("lance_index_job_max_unresolved_per_table", true,
LanceIndexConfigValidator.PositiveLongConfigHandler.class);
assertCallbackWiring("lance_index_job_max_unresolved_per_catalog", true,
LanceIndexConfigValidator.PositiveLongConfigHandler.class);
assertCallbackWiring("lance_index_job_max_unresolved_global", true,
LanceIndexConfigValidator.PositiveLongConfigHandler.class);
assertCallbackWiring("lance_index_max_num_partitions", true,
LanceIndexConfigValidator.PositiveIntConfigHandler.class);
assertCallbackWiring("lance_index_max_num_sub_vectors", true,
Expand All @@ -74,48 +65,6 @@ private static void assertCallbackWiring(String fieldName, boolean masterOnly, C
Assertions.assertEquals(callback, anno.callback(), fieldName);
}

@Test
public void testPositiveLongHandlerAssignsAcceptedValue() throws Exception {
assertLongAssigns("lance_index_job_max_unresolved_per_table");
assertLongAssigns("lance_index_job_max_unresolved_per_catalog");
assertLongAssigns("lance_index_job_max_unresolved_global");
}

private static void assertLongAssigns(String fieldName) throws Exception {
Field field = Config.class.getField(fieldName);
long original = field.getLong(null);
try {
new LanceIndexConfigValidator.PositiveLongConfigHandler().handle(field, " 12345 ");
Assertions.assertEquals(12345L, field.getLong(null), fieldName);
} finally {
field.setLong(null, original);
}
}

@Test
public void testPositiveLongHandlerRejectsInvalidValues() throws Exception {
assertLongRejected("lance_index_job_max_unresolved_per_table", "0");
assertLongRejected("lance_index_job_max_unresolved_per_catalog", "0");
assertLongRejected("lance_index_job_max_unresolved_global", "0");
assertLongRejected("lance_index_job_max_unresolved_per_table", "-8");
assertLongRejected("lance_index_job_max_unresolved_per_table", "not-a-number");
assertLongRejected("lance_index_job_max_unresolved_per_table", "");
}

private static void assertLongRejected(String fieldName, String value) throws Exception {
Field field = Config.class.getField(fieldName);
long original = field.getLong(null);
try {
ConfigException e = Assertions.assertThrows(ConfigException.class,
() -> new LanceIndexConfigValidator.PositiveLongConfigHandler().handle(field, value),
fieldName + " <- \"" + value + "\"");
Assertions.assertTrue(e.getMessage().contains(fieldName), e.getMessage());
Assertions.assertEquals(original, field.getLong(null), fieldName + " must remain unchanged");
} finally {
field.setLong(null, original);
}
}

@Test
public void testPositiveIntHandlerAssignsAcceptedValue() throws Exception {
assertIntAssigns("lance_index_max_num_partitions");
Expand Down Expand Up @@ -167,18 +116,9 @@ public void testSetMutableConfigPath() throws Exception {
tempFile.toFile().deleteOnExit();
config.init(tempFile.toAbsolutePath().toString());

long originalQuota = Config.lance_index_job_max_unresolved_per_catalog;
int originalBound = Config.lance_index_max_num_partitions;
boolean originalGate = Config.enable_lance_index_mutation;
try {
ConfigBase.setMutableConfig("lance_index_job_max_unresolved_per_catalog", "96");
Assertions.assertEquals(96L, Config.lance_index_job_max_unresolved_per_catalog);
Assertions.assertThrows(ConfigException.class,
() -> ConfigBase.setMutableConfig("lance_index_job_max_unresolved_per_catalog", "0"));
Assertions.assertThrows(ConfigException.class,
() -> ConfigBase.setMutableConfig("lance_index_job_max_unresolved_per_catalog", "xyz"));
Assertions.assertEquals(96L, Config.lance_index_job_max_unresolved_per_catalog);

ConfigBase.setMutableConfig("lance_index_max_num_partitions", "8192");
Assertions.assertEquals(8192, Config.lance_index_max_num_partitions);
Assertions.assertThrows(ConfigException.class,
Expand All @@ -188,7 +128,6 @@ public void testSetMutableConfigPath() throws Exception {
ConfigBase.setMutableConfig("enable_lance_index_mutation", "true");
Assertions.assertTrue(Config.enable_lance_index_mutation);
} finally {
Config.lance_index_job_max_unresolved_per_catalog = originalQuota;
Config.lance_index_max_num_partitions = originalBound;
Config.enable_lance_index_mutation = originalGate;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -333,7 +333,6 @@ KEY: 'KEY';
KEYS: 'KEYS';
KILL: 'KILL';
LABEL: 'LABEL';
LANCE: 'LANCE';
LARGEINT: 'LARGEINT';
LAYOUT: 'LAYOUT';
LAST: 'LAST';
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -497,8 +497,6 @@ supportedShowStatement
| SHOW WARM UP JOB wildWhere? #showWarmUpJob
| SHOW PYTHON VERSIONS #showPythonVersions
| SHOW PYTHON PACKAGES IN STRING_LITERAL #showPythonPackages
| SHOW LANCE INDEX JOBS ((FROM | IN) db=multipartIdentifier)? (WHERE expression)? #showLanceIndexJobs
| SHOW LANCE INDEX JOB jobId=INTEGER_VALUE #showLanceIndexJob
;

supportedLoadStatement
Expand Down Expand Up @@ -2204,7 +2202,6 @@ nonReserved
| JSON
| JSONB
| LABEL
| LANCE
| LAST
| LDAP
| LDAP_ADMIN_PASSWORD
Expand Down
25 changes: 0 additions & 25 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,6 @@
import org.apache.doris.datasource.iceberg.IcebergExternalTable;
import org.apache.doris.datasource.iceberg.IcebergSysExternalTable;
import org.apache.doris.datasource.jdbc.JdbcExternalTable;
import org.apache.doris.datasource.lance.job.LanceIndexJobManager;
import org.apache.doris.datasource.paimon.PaimonExternalTable;
import org.apache.doris.datasource.paimon.PaimonSysExternalTable;
import org.apache.doris.deploy.DeployManager;
Expand Down Expand Up @@ -572,8 +571,6 @@ public class Env {

private InsertOverwriteManager insertOverwriteManager;

private LanceIndexJobManager lanceIndexJobManager;

private DNSCache dnsCache;

private final NereidsSqlCacheManager sqlCacheManager;
Expand Down Expand Up @@ -859,7 +856,6 @@ public Env(boolean isCheckpointCatalog) {
this.mtmvService = new MTMVService();
this.eventProcessor = new EventProcessor(mtmvService);
this.insertOverwriteManager = new InsertOverwriteManager();
this.lanceIndexJobManager = new LanceIndexJobManager();
this.dnsCache = new DNSCache();
this.sqlCacheManager = new NereidsSqlCacheManager();
this.sortedPartitionsCacheManager = new NereidsSortedPartitionsCacheManager();
Expand Down Expand Up @@ -985,10 +981,6 @@ public InsertOverwriteManager getInsertOverwriteManager() {
return insertOverwriteManager;
}

public LanceIndexJobManager getLanceIndexJobManager() {
return lanceIndexJobManager;
}

public TabletScheduler getTabletScheduler() {
return tabletScheduler;
}
Expand Down Expand Up @@ -1830,11 +1822,6 @@ private void transferToMaster() {

insertOverwriteManager.allTaskFail();

// A durable RUNNING Lance index job at this point may have lost its result with the
// old master: sweep it to UNKNOWN (and refresh RUNNING back to REQUIRED) before any
// master-only dispatcher could start.
lanceIndexJobManager.onTransferToMaster();

toMasterProgress = "start daemon threads";

// coz current fe was not master fe and didn't get all fes' alive session report before, which cause
Expand Down Expand Up @@ -2685,18 +2672,6 @@ public long saveDictionaryManager(CountingDataOutputStream out, long checksum) t
return checksum;
}

public long loadLanceIndexJobManager(DataInputStream in, long checksum) throws IOException {
this.lanceIndexJobManager = LanceIndexJobManager.read(in);
LOG.info("finished replay lance index job manager from image");
return checksum;
}

public long saveLanceIndexJobManager(CountingDataOutputStream out, long checksum) throws IOException {
this.lanceIndexJobManager.write(out);
LOG.info("finished save lance index job manager to image");
return checksum;
}

// Only called by checkpoint thread
// return the latest image file's absolute path
public String saveImage() throws IOException {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1242,7 +1242,8 @@ public enum ErrorCode {
ERR_LANCE_INDEX_MUTATION_DISABLED(5102, new byte[]{'4', '2', '0', '0', '0'},
"%s is disabled for Lance catalog tables (enable_lance_index_mutation = false)"),

ERR_LANCE_INDEX_JOB_NOT_FOUND(5103, new byte[]{'4', '2', '0', '0', '0'}, "Lance index job not found: %s"),
// 5103 (ERR_LANCE_INDEX_JOB_NOT_FOUND) was removed with the durable Lance index job
// framework; the number stays unused so old client-visible codes keep their meaning.

ERR_NOT_CLOUD_MODE(6000, new byte[]{'4', '2', '0', '0', '0'},
"Command only support in cloud mode.");
Expand Down
Loading
Loading