Schema validation (#557)

See https://iguazio.slack.com/archives/C010QF1UETX/p1625150730019000

pkg/partmgr/partmgr.go

@@ -267,6 +267,21 @@ func (p *PartitionManager) ReadAndUpdateSchema() (err error) {
 	return
 }
 
+func unmarshalSchema(data []byte) (*config.Schema, error) {
+	schemaInstance := &config.Schema{}
+	err := json.Unmarshal(data, schemaInstance)
+	if err != nil {
+		return nil, err
+	}
+	if schemaInstance.TableSchemaInfo.PartitionerInterval == "" {
+		return nil, errors.New("Schema partition interval may not be empty")
+	}
+	if schemaInstance.TableSchemaInfo.ChunckerInterval == "" {
+		return nil, errors.New("Schema chunk interval may not be empty")
+	}
+	return schemaInstance, nil
+}
+
 func (p *PartitionManager) updatePartitionsFromSchema(schema *config.Schema) error {
 	if schema == nil {
 		schemaFilePath := p.GetSchemaFilePath()
@@ -275,8 +290,7 @@ func (p *PartitionManager) updatePartitionsFromSchema(schema *config.Schema) err
 			return errors.Wrapf(innerError, "Failed to read schema at path '%s'.", schemaFilePath)
 		}
 
-		schema = &config.Schema{}
-		innerError = json.Unmarshal(resp.Body(), schema)
+		schema, innerError = unmarshalSchema(resp.Body())
 		if innerError != nil {
 			return errors.Wrapf(innerError, "Failed to unmarshal schema at path '%s'.", schemaFilePath)
 		}

pkg/partmgr/partmgr_test.go

@@ -96,6 +96,20 @@ func TestNewPartitionMngrBadInput(t *testing.T) {
 	assert.Error(t, err)
 }
 
+func TestUnmarshalSchemaError(t *testing.T) {
+	schemaString := []byte(`{}`)
+	_, err := unmarshalSchema(schemaString)
+	assert.Error(t, err)
+
+	schemaString = []byte(`{"tableSchemaInfo": {}}`)
+	_, err = unmarshalSchema(schemaString)
+	assert.Error(t, err)
+
+	schemaString = []byte(`{"tableSchemaInfo": {"PartitionerInterval": ""}}`)
+	_, err = unmarshalSchema(schemaString)
+	assert.Error(t, err)
+}
+
 func TestPartsForRange(tst *testing.T) {
 	numPartitions := 5
 	manager := getPartitionManager(tst)