private SqlIOConfig fetchSystemInfo(String name) {
      String[] sourceComponents = name.split("\\.");
      boolean isTable = isTable(sourceComponents);

      // This source resolver expects sources of format {systemName}.{streamName}[.$table]
      //  * First source part is always system name.
      //  * The last source part could be either a "$table" keyword or stream name. If it is "$table", then stream name
      //    should be the one before the last source part.
      int endIdx = sourceComponents.length - 1;
      int streamIdx = isTable ? endIdx - 1 : endIdx;
      boolean invalidQuery = false;

      if (sourceComponents.length != 2) {
        if (sourceComponents.length != 3 ||
            !sourceComponents[endIdx].equalsIgnoreCase(SAMZA_SQL_QUERY_TABLE_KEYWORD)) {
          invalidQuery = true;
        }
      } else {
        if (sourceComponents[0].equalsIgnoreCase(SAMZA_SQL_QUERY_TABLE_KEYWORD) ||
            sourceComponents[1].equalsIgnoreCase(SAMZA_SQL_QUERY_TABLE_KEYWORD)) {
          invalidQuery = true;
        }
      }

      if (invalidQuery) {
        String msg = String.format("Source %s is not of the format {systemName}.{streamName}[.%s]", name,
            SAMZA_SQL_QUERY_TABLE_KEYWORD);
        LOG.error(msg);
        throw new SamzaException(msg);
      }

      String systemName = sourceComponents[0];
      String streamName = sourceComponents[streamIdx];

      TableDescriptor tableDescriptor = null;
      if (isTable) {
        String tableId = changeLogStorePrefix + "InputTable-" + name.replace(".", "-").replace("$", "-");
        SamzaSqlRelRecordSerdeFactory.SamzaSqlRelRecordSerde keySerde =
            (SamzaSqlRelRecordSerdeFactory.SamzaSqlRelRecordSerde) new SamzaSqlRelRecordSerdeFactory().getSerde(null, null);
        SamzaSqlRelMessageSerdeFactory.SamzaSqlRelMessageSerde valueSerde =
            (SamzaSqlRelMessageSerdeFactory.SamzaSqlRelMessageSerde) new SamzaSqlRelMessageSerdeFactory().getSerde(null, null);
        tableDescriptor = new RocksDbTableDescriptor(tableId, KVSerde.of(keySerde, valueSerde))
            .withChangelogEnabled();
      }

      return new SqlIOConfig(systemName, streamName, fetchSystemConfigs(systemName), tableDescriptor);
    }
Source Resolver implementation that uses static config to return a config corresponding to a system stream.
This Source resolver implementation supports sources of type {systemName}.{streamName}[.$table]
{systemName}.{streamName} indicates a stream
{systemName}.{streamName}.$table indicates a table
samza-sql/src/main/java/org/apache/samza/sql/impl/ConfigBasedIOResolverFactory.java