Skip to content

Minor cleanup in Elasticsearch connector #22016

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
merged 2 commits into from
May 18, 2024
Merged

Minor cleanup in Elasticsearch connector #22016

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6f08311
Convert internal classes to record in Elasticsearch
ebyhr May 17, 2024
a2b7204
Convert IndexMetadata to record in Elasticsearch
ebyhr May 18, 2024
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
112 changes: 31 additions & 81 deletions ...rino-elasticsearch/src/main/java/io/trino/plugin/elasticsearch/ElasticsearchMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,7 +163,7 @@ public ElasticsearchTableHandle getTableHandle(ConnectorSession session, SchemaT
query = Optional.of(parts[1]);
}

if (client.indexExists(table) && !client.getIndexMetadata(table).getSchema().getFields().isEmpty()) {
if (client.indexExists(table) && !client.getIndexMetadata(table).schema().fields().isEmpty()) {
return new ElasticsearchTableHandle(SCAN, schemaName, table, query);
}
}
Expand All @@ -187,7 +187,7 @@ public ConnectorTableMetadata getTableMetadata(ConnectorSession session, Connect
private ConnectorTableMetadata getTableMetadata(String schemaName, String tableName)
{
InternalTableMetadata internalTableMetadata = makeInternalTableMetadata(schemaName, tableName);
return new ConnectorTableMetadata(new SchemaTableName(schemaName, tableName), internalTableMetadata.getColumnMetadata());
return new ConnectorTableMetadata(new SchemaTableName(schemaName, tableName), internalTableMetadata.columnMetadata());
}

private InternalTableMetadata makeInternalTableMetadata(ConnectorTableHandle table)
Expand All @@ -205,12 +205,12 @@ private InternalTableMetadata makeInternalTableMetadata(String schema, String ta

private List<IndexMetadata.Field> getColumnFields(IndexMetadata metadata)
{
Map<String, Long> counts = metadata.getSchema()
.getFields().stream()
.collect(Collectors.groupingBy(f -> f.getName().toLowerCase(ENGLISH), Collectors.counting()));
Map<String, Long> counts = metadata.schema()
.fields().stream()
.collect(Collectors.groupingBy(f -> f.name().toLowerCase(ENGLISH), Collectors.counting()));

return metadata.getSchema().getFields().stream()
.filter(field -> toTrino(field) != null && counts.get(field.getName().toLowerCase(ENGLISH)) <= 1)
return metadata.schema().fields().stream()
.filter(field -> toTrino(field) != null && counts.get(field.name().toLowerCase(ENGLISH)) <= 1)
.collect(toImmutableList());
}

Expand All @@ -224,8 +224,8 @@ private List<ColumnMetadata> makeColumnMetadata(List<IndexMetadata.Field> fields

for (IndexMetadata.Field field : fields) {
result.add(ColumnMetadata.builder()
.setName(field.getName())
.setType(toTrino(field).getType())
.setName(field.name())
.setType(toTrino(field).type())
.build());
}
return result.build();
Expand All @@ -241,11 +241,11 @@ private Map<String, ColumnHandle> makeColumnHandles(List<IndexMetadata.Field> fi

for (IndexMetadata.Field field : fields) {
TypeAndDecoder converted = toTrino(field);
result.put(field.getName(), new ElasticsearchColumnHandle(
field.getName(),
converted.getType(),
converted.getDecoderDescriptor(),
supportsPredicates(field.getType())));
result.put(field.name(), new ElasticsearchColumnHandle(
field.name(),
converted.type(),
converted.decoderDescriptor(),
supportsPredicates(field.type())));
}

return result.buildOrThrow();
Expand All @@ -258,7 +258,7 @@ private static boolean supportsPredicates(IndexMetadata.Type type)
}

if (type instanceof PrimitiveType) {
switch (((PrimitiveType) type).getName().toLowerCase(ENGLISH)) {
switch (((PrimitiveType) type).name().toLowerCase(ENGLISH)) {
case "boolean":
case "byte":
case "short":
Expand All @@ -281,7 +281,7 @@ private TypeAndDecoder toTrino(IndexMetadata.Field field)

private TypeAndDecoder toTrino(String prefix, IndexMetadata.Field field)
{
String path = appendPath(prefix, field.getName());
String path = appendPath(prefix, field.name());

checkArgument(!field.asRawJson() || !field.isArray(), format("A column, (%s) cannot be declared as a Trino array and also be rendered as json.", path));

Expand All @@ -291,12 +291,12 @@ private TypeAndDecoder toTrino(String prefix, IndexMetadata.Field field)

if (field.isArray()) {
TypeAndDecoder element = toTrino(path, elementField(field));
return new TypeAndDecoder(new ArrayType(element.getType()), new ArrayDecoder.Descriptor(element.getDecoderDescriptor()));
return new TypeAndDecoder(new ArrayType(element.type()), new ArrayDecoder.Descriptor(element.decoderDescriptor()));
}

IndexMetadata.Type type = field.getType();
IndexMetadata.Type type = field.type();
if (type instanceof PrimitiveType primitiveType) {
switch (primitiveType.getName()) {
switch (primitiveType.name()) {
case "float":
return new TypeAndDecoder(REAL, new RealDecoder.Descriptor(path));
case "double":
Expand Down Expand Up @@ -324,21 +324,21 @@ else if (type instanceof ScaledFloatType) {
return new TypeAndDecoder(DOUBLE, new DoubleDecoder.Descriptor(path));
}
else if (type instanceof DateTimeType dateTimeType) {
if (dateTimeType.getFormats().isEmpty()) {
if (dateTimeType.formats().isEmpty()) {
return new TypeAndDecoder(TIMESTAMP_MILLIS, new TimestampDecoder.Descriptor(path));
}
// otherwise, skip -- we don't support custom formats, yet
}
else if (type instanceof ObjectType objectType) {
ImmutableList.Builder<RowType.Field> rowFieldsBuilder = ImmutableList.builder();
ImmutableList.Builder<RowDecoder.NameAndDescriptor> decoderFields = ImmutableList.builder();
for (IndexMetadata.Field rowField : objectType.getFields()) {
String name = rowField.getName();
for (IndexMetadata.Field rowField : objectType.fields()) {
String name = rowField.name();
TypeAndDecoder child = toTrino(path, rowField);

if (child != null) {
decoderFields.add(new RowDecoder.NameAndDescriptor(name, child.getDecoderDescriptor()));
rowFieldsBuilder.add(RowType.field(name, child.getType()));
decoderFields.add(new RowDecoder.NameAndDescriptor(name, child.decoderDescriptor()));
rowFieldsBuilder.add(RowType.field(name, child.type()));
}
}

Expand All @@ -365,7 +365,7 @@ private static String appendPath(String base, String element)
public static IndexMetadata.Field elementField(IndexMetadata.Field field)
{
checkArgument(field.isArray(), "Cannot get element field from a non-array field");
return new IndexMetadata.Field(field.asRawJson(), false, field.getName(), field.getType());
return new IndexMetadata.Field(field.asRawJson(), false, field.name(), field.type());
}

@Override
Expand Down Expand Up @@ -402,7 +402,7 @@ public Map<String, ColumnHandle> getColumnHandles(ConnectorSession session, Conn
}

InternalTableMetadata tableMetadata = makeInternalTableMetadata(tableHandle);
return tableMetadata.getColumnHandles();
return tableMetadata.columnHandles();
}

@Override
Expand Down Expand Up @@ -547,9 +547,9 @@ public Optional<ConstraintApplicationResult<ConnectorTableHandle>> applyFilter(C

if (!newRegexes.containsKey(columnName) && pattern instanceof Slice) {
IndexMetadata metadata = client.getIndexMetadata(handle.index());
if (metadata.getSchema()
.getFields().stream()
.anyMatch(field -> columnName.equals(field.getName()) && field.getType() instanceof PrimitiveType && "keyword".equals(((PrimitiveType) field.getType()).getName()))) {
if (metadata.schema()
.fields().stream()
.anyMatch(field -> columnName.equals(field.name()) && field.type() instanceof PrimitiveType && "keyword".equals(((PrimitiveType) field.type()).name()))) {
newRegexes.put(columnName, likeToRegexp((Slice) pattern, escape));
continue;
}
Expand Down Expand Up @@ -673,57 +673,7 @@ public Optional<TableFunctionApplicationResult<ConnectorTableHandle>> applyTable
return Optional.of(new TableFunctionApplicationResult<>(tableHandle, columnHandles));
}

private static class InternalTableMetadata
{
private final SchemaTableName tableName;
private final List<ColumnMetadata> columnMetadata;
private final Map<String, ColumnHandle> columnHandles;

public InternalTableMetadata(
SchemaTableName tableName,
List<ColumnMetadata> columnMetadata,
Map<String, ColumnHandle> columnHandles)
{
this.tableName = tableName;
this.columnMetadata = columnMetadata;
this.columnHandles = columnHandles;
}

public SchemaTableName getTableName()
{
return tableName;
}

public List<ColumnMetadata> getColumnMetadata()
{
return columnMetadata;
}

public Map<String, ColumnHandle> getColumnHandles()
{
return columnHandles;
}
}

private static class TypeAndDecoder
{
private final Type type;
private final DecoderDescriptor decoderDescriptor;

public TypeAndDecoder(Type type, DecoderDescriptor decoderDescriptor)
{
this.type = type;
this.decoderDescriptor = decoderDescriptor;
}
private record InternalTableMetadata(SchemaTableName tableName, List<ColumnMetadata> columnMetadata, Map<String, ColumnHandle> columnHandles) {}

public Type getType()
{
return type;
}

public DecoderDescriptor getDecoderDescriptor()
{
return decoderDescriptor;
}
}
private record TypeAndDecoder(Type type, DecoderDescriptor decoderDescriptor) {}
}
2 changes: 1 addition & 1 deletion ...elasticsearch/src/main/java/io/trino/plugin/elasticsearch/client/ElasticsearchClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -391,7 +391,7 @@ public List<String> getIndexes()
if (docsCount == 0 && deletedDocsCount == 0) {
try {
// without documents, the index won't have any dynamic mappings, but maybe there are some explicit ones
if (getIndexMetadata(index).getSchema().getFields().isEmpty()) {
if (getIndexMetadata(index).schema().fields().isEmpty()) {
continue;
}
}
Expand Down
102 changes: 17 additions & 85 deletions ...trino-elasticsearch/src/main/java/io/trino/plugin/elasticsearch/client/IndexMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,125 +21,57 @@
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;

public class IndexMetadata
public record IndexMetadata(ObjectType schema)
{
private final ObjectType schema;

public IndexMetadata(ObjectType schema)
public IndexMetadata
{
this.schema = requireNonNull(schema, "schema is null");
requireNonNull(schema, "schema is null");
}

public ObjectType getSchema()
public record Field(boolean asRawJson, boolean isArray, String name, Type type)
{
return schema;
}

public static class Field
{
private final boolean asRawJson;
private final boolean isArray;
private final String name;
private final Type type;

public Field(boolean asRawJson, boolean isArray, String name, Type type)
public Field
{
checkArgument(!asRawJson || !isArray,
format("A column, (%s) cannot be declared as a Trino array and also be rendered as json.", name));
this.asRawJson = asRawJson;
this.isArray = isArray;
this.name = requireNonNull(name, "name is null");
this.type = requireNonNull(type, "type is null");
}

public boolean asRawJson()
{
return asRawJson;
}

public boolean isArray()
{
return isArray;
}

public String getName()
{
return name;
}

public Type getType()
{
return type;
requireNonNull(name, "name is null");
requireNonNull(type, "type is null");
}
}

public interface Type {}

public static class PrimitiveType
public record PrimitiveType(String name)
implements Type
{
private final String name;

public PrimitiveType(String name)
public PrimitiveType
{
this.name = requireNonNull(name, "name is null");
}

public String getName()
{
return name;
requireNonNull(name, "name is null");
}
}

public static class DateTimeType
public record DateTimeType(List<String> formats)
implements Type
{
private final List<String> formats;

public DateTimeType(List<String> formats)
public DateTimeType
{
requireNonNull(formats, "formats is null");

this.formats = ImmutableList.copyOf(formats);
}

public List<String> getFormats()
{
return formats;
formats = ImmutableList.copyOf(formats);
}
}

public static class ObjectType
public record ObjectType(List<Field> fields)
implements Type
{
private final List<Field> fields;

public ObjectType(List<Field> fields)
public ObjectType
{
requireNonNull(fields, "fields is null");

this.fields = ImmutableList.copyOf(fields);
}

public List<Field> getFields()
{
return fields;
fields = ImmutableList.copyOf(fields);
}
}

public static class ScaledFloatType
public record ScaledFloatType(double scale)
implements Type
{
private final double scale;

public ScaledFloatType(double scale)
{
this.scale = scale;
}

public double getScale()
{
return scale;
}
}
}