MQTT Query Interface

.github/workflows/mqttClientTest.yml

@@ -0,0 +1,51 @@
+name: Polypheny-DB MQTT Client Test
+
+on:
+    push: { branches: mqtt-interface }
+
+jobs:
+    build:
+        runs-on: ubuntu-latest
+        name: MQTT Client Tests (Java 17)
+        steps:
+            -   name: Checkout
+                uses: actions/checkout@v2
+            -   name: Set up JDK
+                uses: actions/setup-java@v1
+                with:
+                    java-version: 17
+            -   name: Set env variable
+                run: |
+                    echo "POLYPHENY_HOME=$GITHUB_WORKSPACE" >> $GITHUB_ENV
+            -   name: Create folders for certs
+                run: |
+                    mkdir $POLYPHENY_HOME/.polypheny
+                    mkdir $POLYPHENY_HOME/.polypheny/certs
+
+            -   name: Start Mosquitto broker in docker container
+                uses: namoshek/mosquitto-github-action@v1
+                with:
+                    version: '1.6'
+                    ports: '1883:1883'
+                    container-name: 'mqttBroker'
+
+            -   name: Assemble
+                uses: nick-invision/retry@v2
+                with:
+                    max_attempts: 2
+                    timeout_minutes: 60
+                    command: ./gradlew assemble
+
+            - name: Build Plugins
+              uses: nick-invision/retry@v2
+              with:
+                  max_attempts: 1
+                  timeout_minutes: 60
+                  command: ./gradlew assemblePlugins
+
+            -   name: Execute MQTT client tests
+                uses: nick-invision/retry@v2
+                with:
+                    max_attempts: 1
+                    timeout_minutes: 30
+                    command: ./gradlew mqttTests

core/src/main/java/org/polypheny/db/adapter/Adapter.java

@@ -53,6 +53,7 @@
 import org.polypheny.db.catalog.entity.CatalogColumnPlacement;
 import org.polypheny.db.catalog.entity.CatalogPartitionPlacement;
 import org.polypheny.db.catalog.entity.CatalogTable;
+import org.polypheny.db.catalog.exceptions.NoTablePrimaryKeyException;
 import org.polypheny.db.config.Config;
 import org.polypheny.db.config.Config.ConfigListener;
 import org.polypheny.db.config.ConfigDocker;
@@ -406,7 +407,13 @@ public void updateSettings( Map<String, String> newSettings ) {
         this.validateSettings( newSettings, false );
         List<String> updatedSettings = this.applySettings( newSettings );
         this.reloadSettings( updatedSettings );
-        Catalog.getInstance().updateAdapterSettings( getAdapterId(), newSettings );
+        Catalog catalog = Catalog.getInstance();
+        catalog.updateAdapterSettings( getAdapterId(), getCurrentSettings() );
+        try {
+            catalog.commit();
+        } catch ( NoTablePrimaryKeyException e ) {
+            throw new RuntimeException( e );
+        }
     }
 
 

core/src/main/java/org/polypheny/db/adapter/enumerable/EnumerableDocumentValues.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2019-2023 The Polypheny Project
+ *
+ * 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 org.polypheny.db.adapter.enumerable;
+
+import com.google.common.collect.ImmutableList;
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.calcite.linq4j.tree.BlockBuilder;
+import org.apache.calcite.linq4j.tree.Expression;
+import org.apache.calcite.linq4j.tree.Expressions;
+import org.apache.calcite.linq4j.tree.Primitive;
+import org.bson.BsonValue;
+import org.polypheny.db.adapter.java.JavaTypeFactory;
+import org.polypheny.db.algebra.core.document.DocumentValues;
+import org.polypheny.db.algebra.type.AlgDataType;
+import org.polypheny.db.algebra.type.AlgDataTypeField;
+import org.polypheny.db.plan.AlgOptCluster;
+import org.polypheny.db.plan.AlgTraitSet;
+import org.polypheny.db.schema.ModelTrait;
+import org.polypheny.db.util.BuiltInMethod;
+
+public class EnumerableDocumentValues extends DocumentValues implements EnumerableAlg {
+
+    /**
+     * Creates a {@link DocumentValues}.
+     * {@link ModelTrait#DOCUMENT} node, which contains values.
+     *
+     * @param cluster
+     * @param traitSet
+     * @param rowType
+     * @param documentTuples
+     */
+    public EnumerableDocumentValues( AlgOptCluster cluster, AlgTraitSet traitSet, AlgDataType rowType, List<BsonValue> documentTuples ) {
+        super( cluster, traitSet, rowType, ImmutableList.copyOf( documentTuples ) );
+    }
+
+
+    @Override
+    public Result implement( EnumerableAlgImplementor implementor, Prefer pref ) {
+        final JavaTypeFactory typeFactory = (JavaTypeFactory) getCluster().getTypeFactory();
+        final BlockBuilder builder = new BlockBuilder();
+        final PhysType physType =
+                PhysTypeImpl.of(
+                        implementor.getTypeFactory(),
+                        getRowType(),
+                        pref.preferCustom() );
+        final Type rowClass = physType.getJavaRowType();
+
+        final List<Expression> expressions = new ArrayList<>();
+        final List<AlgDataTypeField> fields = rowType.getFieldList();
+        for ( BsonValue doc : documentTuples ) {
+            final List<Expression> literals = new ArrayList<>();
+
+            literals.add( Expressions.constant( doc.asDocument().toJson() ) );
+
+            expressions.add( physType.record( literals ) );
+        }
+        builder.add(
+                Expressions.return_(
+                        null,
+                        Expressions.call(
+                                BuiltInMethod.AS_ENUMERABLE.method,
+                                Expressions.newArrayInit( Primitive.box( rowClass ), expressions ) ) ) );
+        return implementor.result( physType, builder.toBlock() );
+    }
+
+}

core/src/main/java/org/polypheny/db/algebra/logical/document/LogicalDocumentValues.java

@@ -38,6 +38,7 @@
 import org.polypheny.db.plan.AlgTraitSet;
 import org.polypheny.db.plan.Convention;
 import org.polypheny.db.rex.RexLiteral;
+import org.polypheny.db.schema.ModelTrait;
 import org.polypheny.db.type.PolyType;
 import org.polypheny.db.type.PolyTypeFactoryImpl;
 import org.polypheny.db.util.BsonUtil;
@@ -77,7 +78,7 @@ public class LogicalDocumentValues extends DocumentValues implements RelationalT
      * @param tuples the documents in their native BSON format
      */
     public LogicalDocumentValues( AlgOptCluster cluster, AlgDataType defaultRowType, AlgTraitSet traitSet, ImmutableList<BsonValue> tuples ) {
-        super( cluster, traitSet, defaultRowType, tuples );
+        super( cluster, traitSet.replace( ModelTrait.DOCUMENT ), defaultRowType, tuples );
     }
 
 

core/src/main/java/org/polypheny/db/algebra/rules/DocumentToEnumerableRule.java

@@ -19,6 +19,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.polypheny.db.adapter.enumerable.EnumerableAggregate;
 import org.polypheny.db.adapter.enumerable.EnumerableConvention;
+import org.polypheny.db.adapter.enumerable.EnumerableDocumentValues;
 import org.polypheny.db.adapter.enumerable.EnumerableFilter;
 import org.polypheny.db.adapter.enumerable.EnumerableLimit;
 import org.polypheny.db.adapter.enumerable.EnumerableProject;
@@ -30,6 +31,7 @@
 import org.polypheny.db.algebra.logical.document.LogicalDocumentFilter;
 import org.polypheny.db.algebra.logical.document.LogicalDocumentProject;
 import org.polypheny.db.algebra.logical.document.LogicalDocumentSort;
+import org.polypheny.db.algebra.logical.document.LogicalDocumentValues;
 import org.polypheny.db.plan.AlgOptRule;
 import org.polypheny.db.plan.AlgOptRuleCall;
 import org.polypheny.db.plan.AlgOptRuleOperand;
@@ -43,6 +45,8 @@ public class DocumentToEnumerableRule extends AlgOptRule {
     public static DocumentToEnumerableRule FILTER_TO_ENUMERABLE = new DocumentToEnumerableRule( Type.FILTER, operand( LogicalDocumentFilter.class, any() ), "DOCUMENT_FILTER_TO_ENUMERABLE" );
     public static DocumentToEnumerableRule SORT_TO_ENUMERABLE = new DocumentToEnumerableRule( Type.SORT, operand( LogicalDocumentSort.class, any() ), "DOCUMENT_SORT_TO_ENUMERABLE" );
 
+    public static DocumentToEnumerableRule VALUES_TO_ENUMERABLE = new DocumentToEnumerableRule( Type.VALUES, operand( LogicalDocumentValues.class, any() ), "DOCUMENT_VALUES_TO_ENUMERABLE" );
+
     private final Type type;
 
 
@@ -62,13 +66,26 @@ public void onMatch( AlgOptRuleCall call ) {
             convertAggregate( call );
         } else if ( type == Type.SORT ) {
             convertSort( call );
+        } else if ( type == Type.VALUES ) {
+            convertValues( call );
         } else {
             throw new UnsupportedOperationException( "This document is not supported." );
         }
 
     }
 
 
+    private void convertValues( AlgOptRuleCall call ) {
+        LogicalDocumentValues values = call.alg( 0 );
+        AlgTraitSet out = values.getTraitSet().replace( EnumerableConvention.INSTANCE );
+
+        EnumerableDocumentValues enumerable = new EnumerableDocumentValues( values.getCluster(), out, values.getRowType(), values.documentTuples );
+        call.transformTo( enumerable );
+
+        // call.transformTo( values.getRelationalEquivalent() );
+    }
+
+
     private void convertSort( AlgOptRuleCall call ) {
         LogicalDocumentSort sort = call.alg( 0 );
         AlgTraitSet out = sort.getTraitSet().replace( EnumerableConvention.INSTANCE );

core/src/main/java/org/polypheny/db/catalog/Catalog.java

@@ -1233,6 +1233,14 @@ protected final boolean isValidIdentifier( final String str ) {
      */
     public abstract void deleteQueryInterface( int ifaceId );
 
+    /**
+     * Update settings of a query interface
+     *
+     * @param queryInterfaceId The id of the query interface
+     * @param newSettings The new settings for the query interface
+     */
+    public abstract void updateQueryInterfaceSettings( int queryInterfaceId, Map<String, String> newSettings );
+
     /**
      * Adds a partition to the catalog
      *
@@ -1926,7 +1934,7 @@ public enum EntityType {
         SOURCE( 2 ),
         VIEW( 3 ),
         MATERIALIZED_VIEW( 4 );
-        // STREAM, ...
+        //STREAM(5);
 
         private final int id;
 

core/src/main/java/org/polypheny/db/catalog/entity/CatalogCollection.java

@@ -61,13 +61,13 @@ public Serializable[] getParameterArray() {
     public CatalogCollection addPlacement( int adapterId ) {
         List<Integer> placements = new ArrayList<>( this.placements );
         placements.add( adapterId );
-        return new CatalogCollection( databaseId, namespaceId, id, name, placements, EntityType.ENTITY, physicalName );
+        return new CatalogCollection( databaseId, namespaceId, id, name, placements, entityType, physicalName );
     }
 
 
     public CatalogCollection removePlacement( int adapterId ) {
         List<Integer> placements = this.placements.stream().filter( id -> id != adapterId ).collect( Collectors.toList() );
-        return new CatalogCollection( databaseId, namespaceId, id, name, placements, EntityType.ENTITY, physicalName );
+        return new CatalogCollection( databaseId, namespaceId, id, name, placements, entityType, physicalName );
     }
 
 

core/src/main/java/org/polypheny/db/config/RuntimeConfig.java

@@ -159,21 +159,21 @@ public enum RuntimeConfig {
     DYNAMIC_QUERYING(
             "statistics/useDynamicQuerying",
             "Use statistics for query assistance.",
-            true,
+            false,
             ConfigType.BOOLEAN,
             "statisticSettingsGroup" ),
 
     STATISTICS_ON_STARTUP(
             "statistics/statisticsOnStartup",
             "Whether to build statistics for all stored data on system startup.",
-            true,
+            false,
             ConfigType.BOOLEAN,
             "statisticSettingsGroup" ),
 
     ACTIVE_TRACKING(
             "statistics/activeTracking",
             "All transactions are tracked and statistics collected during execution.",
-            true,
+            false,
             ConfigType.BOOLEAN,
             "statisticSettingsGroup" ),
 
@@ -418,7 +418,7 @@ public enum RuntimeConfig {
     MONITORING_QUEUE_ACTIVE(
             "runtime/monitoringQueueActive",
             "Enables automatic monitoring of executed events in workload monitoring. If disabled no events are captured, hence the queue remains empty. This also effects routing!",
-            true,
+            false,
             ConfigType.BOOLEAN,
             "monitoringSettingsQueueGroup" ),
 

core/src/main/java/org/polypheny/db/iface/QueryInterface.java

@@ -20,13 +20,16 @@
 import java.beans.PropertyChangeEvent;
 import java.beans.PropertyChangeListener;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import org.pf4j.ExtensionPoint;
+import org.polypheny.db.catalog.Catalog;
+import org.polypheny.db.catalog.exceptions.NoTablePrimaryKeyException;
 import org.polypheny.db.languages.LanguageManager;
 import org.polypheny.db.transaction.TransactionManager;
 
@@ -61,10 +64,15 @@ public QueryInterface(
         this.authenticator = authenticator;
         this.queryInterfaceId = queryInterfaceId;
         this.uniqueName = uniqueName;
-        this.settings = settings;
+        //this.settings = settings;
         this.supportsDml = supportsDml;
         this.supportsDdl = supportsDdl;
 
+        this.settings = new HashMap<>(settings.size());
+        for ( Map.Entry<String, String> entry : settings.entrySet()) {
+            this.settings.put(entry.getKey(), entry.getValue());
+        }
+
         LanguageManager.getINSTANCE().addObserver( this );
     }
 
@@ -120,6 +128,13 @@ public void updateSettings( Map<String, String> newSettings ) {
         this.validateSettings( newSettings, false );
         List<String> updatedSettings = this.applySettings( newSettings );
         this.reloadSettings( updatedSettings );
+        Catalog catalog = Catalog.getInstance();
+        Catalog.getInstance().updateQueryInterfaceSettings( getQueryInterfaceId(), getCurrentSettings() );
+        try {
+            catalog.commit();
+        } catch ( NoTablePrimaryKeyException e ) {
+            throw new RuntimeException( e );
+        }
     }
 
 

core/src/main/java/org/polypheny/db/plan/volcano/VolcanoPlanner.java

@@ -830,6 +830,7 @@ public void registerModelRules() {
         addRule( DocumentToEnumerableRule.FILTER_TO_ENUMERABLE );
         addRule( DocumentToEnumerableRule.AGGREGATE_TO_ENUMERABLE );
         addRule( DocumentToEnumerableRule.SORT_TO_ENUMERABLE );
+        addRule( DocumentToEnumerableRule.VALUES_TO_ENUMERABLE );
 
         // Relational
     }

core/src/main/java/org/polypheny/db/processing/LogicalAlgAnalyzeShuttle.java

@@ -38,6 +38,7 @@
 import org.polypheny.db.algebra.logical.document.LogicalDocumentScan;
 import org.polypheny.db.algebra.logical.document.LogicalDocumentSort;
 import org.polypheny.db.algebra.logical.document.LogicalDocumentTransformer;
+import org.polypheny.db.algebra.logical.document.LogicalDocumentValues;
 import org.polypheny.db.algebra.logical.lpg.LogicalGraph;
 import org.polypheny.db.algebra.logical.lpg.LogicalLpgAggregate;
 import org.polypheny.db.algebra.logical.lpg.LogicalLpgFilter;
@@ -261,6 +262,11 @@ public AlgNode visit( LogicalDocumentScan scan ) {
         return super.visit( scan );
     }
 
+    public AlgNode visit( LogicalDocumentValues values ) {
+        hashBasis.add( "LogicalDocumentValues" );
+        return super.visit( values );
+    }
+
 
     @Override
     public AlgNode visit( LogicalDocumentSort sort ) {

core/src/main/java/org/polypheny/db/runtime/functions/MqlFunctions.java

@@ -781,6 +781,8 @@ private static Object transformBsonToPrimitive( BsonValue doc ) {
                 return doc.asDocument().entrySet().stream().collect( Collectors.toMap( Entry::getKey, e -> transformBsonToPrimitive( e.getValue() ) ) );
             case ARRAY:
                 return doc.asArray().stream().map( MqlFunctions::transformBsonToPrimitive ).collect( Collectors.toList() );
+            case BOOLEAN:
+                return doc.asBoolean().getValue();
             default:
                 return null;
         }